Merge pull request #18917 from edx/arch/user-authn-app

Consolidate user login and authentication code
This commit is contained in:
Nimisha Asthagiri
2018-09-15 10:00:51 -04:00
committed by GitHub
50 changed files with 2421 additions and 2416 deletions

View File

@@ -17,10 +17,10 @@ import third_party_auth
from course_modes.models import CourseMode
from email_marketing.models import EmailMarketingConfiguration
from lms.djangoapps.email_marketing.tasks import get_email_cookies_via_sailthru, update_user, update_user_email
from openedx.core.djangoapps.user_authn.cookies import CREATE_LOGON_COOKIE
from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY
from openedx.core.djangoapps.user_api.accounts.signals import USER_RETIRE_THIRD_PARTY_MAILINGS
from openedx.core.djangoapps.waffle_utils import WaffleSwitchNamespace
from student.cookies import CREATE_LOGON_COOKIE
from student.signals import SAILTHRU_AUDIT_PURCHASE
from student.views import REGISTER_USER
from util.model_utils import USER_FIELD_CHANGED

View File

@@ -1,1142 +0,0 @@
# -*- coding: utf-8 -*-
""" Tests for student account views. """
import logging
import re
from unittest import skipUnless
from urllib import urlencode
import ddt
import mock
from django.conf import settings
from django.contrib import messages
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.contrib.messages.middleware import MessageMiddleware
from django.contrib.sessions.middleware import SessionMiddleware
from django.core import mail
from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
from django.http import HttpRequest
from django.test import TestCase
from django.test.client import RequestFactory
from django.test.utils import override_settings
from django.utils.translation import ugettext as _
from edx_oauth2_provider.tests.factories import AccessTokenFactory, ClientFactory, RefreshTokenFactory
from edx_rest_api_client import exceptions
from http.cookies import SimpleCookie
from oauth2_provider.models import AccessToken as dot_access_token
from oauth2_provider.models import RefreshToken as dot_refresh_token
from provider.oauth2.models import AccessToken as dop_access_token
from provider.oauth2.models import RefreshToken as dop_refresh_token
from testfixtures import LogCapture
from course_modes.models import CourseMode
from lms.djangoapps.commerce.models import CommerceConfiguration
from lms.djangoapps.commerce.tests import factories
from lms.djangoapps.commerce.tests.mocks import mock_get_orders
from lms.djangoapps.student_account.views import login_and_registration_form
from openedx.core.djangoapps.oauth_dispatch.tests import factories as dot_factories
from openedx.core.djangoapps.programs.tests.mixins import ProgramsApiConfigMixin
from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
from openedx.core.djangoapps.site_configuration.tests.mixins import SiteMixin
from openedx.core.djangoapps.theming.tests.test_util import with_comprehensive_theme_context
from openedx.core.djangoapps.user_api.accounts.api import activate_account, create_account
from openedx.core.djangolib.js_utils import dump_js_escaped_json
from openedx.core.djangolib.markup import HTML, Text
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
from student.tests.factories import UserFactory
from student_account.views import account_settings_context, get_user_orders
from third_party_auth.tests.testutil import ThirdPartyAuthTestMixin, simulate_running_pipeline
from util.testing import UrlResetMixin
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from openedx.core.djangoapps.user_api.errors import UserAPIInternalError
LOGGER_NAME = 'audit'
User = get_user_model() # pylint:disable=invalid-name
FEATURES_WITH_FAILED_PASSWORD_RESET_EMAIL = settings.FEATURES.copy()
FEATURES_WITH_FAILED_PASSWORD_RESET_EMAIL['ENABLE_PASSWORD_RESET_FAILURE_EMAIL'] = True
@ddt.ddt
class StudentAccountUpdateTest(CacheIsolationTestCase, UrlResetMixin):
""" Tests for the student account views that update the user's account information. """
USERNAME = u"heisenberg"
ALTERNATE_USERNAME = u"walt"
OLD_PASSWORD = u"ḅḷüëṡḳÿ"
NEW_PASSWORD = u"🄱🄸🄶🄱🄻🅄🄴"
OLD_EMAIL = u"walter@graymattertech.com"
NEW_EMAIL = u"walt@savewalterwhite.com"
INVALID_ATTEMPTS = 100
INVALID_KEY = u"123abc"
URLCONF_MODULES = ['student_accounts.urls']
ENABLED_CACHES = ['default']
def setUp(self):
super(StudentAccountUpdateTest, self).setUp()
# Create/activate a new account
activation_key = create_account(self.USERNAME, self.OLD_PASSWORD, self.OLD_EMAIL)
activate_account(activation_key)
# Login
result = self.client.login(username=self.USERNAME, password=self.OLD_PASSWORD)
self.assertTrue(result)
@skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in LMS')
def test_password_change(self):
# Request a password change while logged in, simulating
# use of the password reset link from the account page
response = self._change_password()
self.assertEqual(response.status_code, 200)
# Check that an email was sent
self.assertEqual(len(mail.outbox), 1)
# Retrieve the activation link from the email body
email_body = mail.outbox[0].body
result = re.search(r'(?P<url>https?://[^\s]+)', email_body)
self.assertIsNot(result, None)
activation_link = result.group('url')
# Visit the activation link
response = self.client.get(activation_link)
self.assertEqual(response.status_code, 200)
# Submit a new password and follow the redirect to the success page
response = self.client.post(
activation_link,
# These keys are from the form on the current password reset confirmation page.
{'new_password1': self.NEW_PASSWORD, 'new_password2': self.NEW_PASSWORD},
follow=True
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Your password has been reset.")
# Log the user out to clear session data
self.client.logout()
# Verify that the new password can be used to log in
result = self.client.login(username=self.USERNAME, password=self.NEW_PASSWORD)
self.assertTrue(result)
# Try reusing the activation link to change the password again
# Visit the activation link again.
response = self.client.get(activation_link)
self.assertEqual(response.status_code, 200)
self.assertContains(response, "This password reset link is invalid. It may have been used already.")
self.client.logout()
# Verify that the old password cannot be used to log in
result = self.client.login(username=self.USERNAME, password=self.OLD_PASSWORD)
self.assertFalse(result)
# Verify that the new password continues to be valid
result = self.client.login(username=self.USERNAME, password=self.NEW_PASSWORD)
self.assertTrue(result)
def test_password_change_failure(self):
with mock.patch('openedx.core.djangoapps.user_api.accounts.api.request_password_change',
side_effect=UserAPIInternalError):
self._change_password()
self.assertRaises(UserAPIInternalError)
@override_settings(FEATURES=FEATURES_WITH_FAILED_PASSWORD_RESET_EMAIL)
def test_password_reset_failure_email(self):
"""Test that a password reset failure email notification is sent, when enabled."""
# Log the user out
self.client.logout()
bad_email = 'doesnotexist@example.com'
response = self._change_password(email=bad_email)
self.assertEqual(response.status_code, 200)
# Check that an email was sent
self.assertEqual(len(mail.outbox), 1)
# Verify that the body contains the failed password reset message
sent_message = mail.outbox[0]
text_body = sent_message.body
html_body = sent_message.alternatives[0][0]
for email_body in [text_body, html_body]:
msg = 'However, there is currently no user account associated with your email address: {email}'.format(
email=bad_email
)
assert u'reset for your user account at {}'.format(settings.PLATFORM_NAME) in email_body
assert 'password_reset_confirm' not in email_body, 'The link should not be added if user was not found'
assert msg in email_body
@ddt.data(True, False)
def test_password_change_logged_out(self, send_email):
# Log the user out
self.client.logout()
# Request a password change while logged out, simulating
# use of the password reset link from the login page
if send_email:
response = self._change_password(email=self.OLD_EMAIL)
self.assertEqual(response.status_code, 200)
else:
# Don't send an email in the POST data, simulating
# its (potentially accidental) omission in the POST
# data sent from the login page
response = self._change_password()
self.assertEqual(response.status_code, 400)
def test_access_token_invalidation_logged_out(self):
self.client.logout()
user = User.objects.get(email=self.OLD_EMAIL)
self._create_dop_tokens(user)
self._create_dot_tokens(user)
response = self._change_password(email=self.OLD_EMAIL)
self.assertEqual(response.status_code, 200)
self.assert_access_token_destroyed(user)
def test_access_token_invalidation_logged_in(self):
user = User.objects.get(email=self.OLD_EMAIL)
self._create_dop_tokens(user)
self._create_dot_tokens(user)
response = self._change_password()
self.assertEqual(response.status_code, 200)
self.assert_access_token_destroyed(user)
def test_password_change_inactive_user(self):
# Log out the user created during test setup
self.client.logout()
# Create a second user, but do not activate it
create_account(self.ALTERNATE_USERNAME, self.OLD_PASSWORD, self.NEW_EMAIL)
# Send the view the email address tied to the inactive user
response = self._change_password(email=self.NEW_EMAIL)
# Expect that the activation email is still sent,
# since the user may have lost the original activation email.
self.assertEqual(response.status_code, 200)
self.assertEqual(len(mail.outbox), 1)
def test_password_change_no_user(self):
# Log out the user created during test setup
self.client.logout()
with LogCapture(LOGGER_NAME, level=logging.INFO) as logger:
# Send the view an email address not tied to any user
response = self._change_password(email=self.NEW_EMAIL)
self.assertEqual(response.status_code, 200)
logger.check((LOGGER_NAME, 'INFO', 'Invalid password reset attempt'))
def test_password_change_rate_limited(self):
# Log out the user created during test setup, to prevent the view from
# selecting the logged-in user's email address over the email provided
# in the POST data
self.client.logout()
# Make many consecutive bad requests in an attempt to trigger the rate limiter
for __ in xrange(self.INVALID_ATTEMPTS):
self._change_password(email=self.NEW_EMAIL)
response = self._change_password(email=self.NEW_EMAIL)
self.assertEqual(response.status_code, 403)
@ddt.data(
('post', 'password_change_request', []),
)
@ddt.unpack
def test_require_http_method(self, correct_method, url_name, args):
wrong_methods = {'get', 'put', 'post', 'head', 'options', 'delete'} - {correct_method}
url = reverse(url_name, args=args)
for method in wrong_methods:
response = getattr(self.client, method)(url)
self.assertEqual(response.status_code, 405)
def _change_password(self, email=None):
"""Request to change the user's password. """
data = {}
if email:
data['email'] = email
return self.client.post(path=reverse('password_change_request'), data=data)
def _create_dop_tokens(self, user=None):
"""Create dop access token for given user if user provided else for default user."""
if not user:
user = User.objects.get(email=self.OLD_EMAIL)
client = ClientFactory()
access_token = AccessTokenFactory(user=user, client=client)
RefreshTokenFactory(user=user, client=client, access_token=access_token)
def _create_dot_tokens(self, user=None):
"""Create dop access token for given user if user provided else for default user."""
if not user:
user = User.objects.get(email=self.OLD_EMAIL)
application = dot_factories.ApplicationFactory(user=user)
access_token = dot_factories.AccessTokenFactory(user=user, application=application)
dot_factories.RefreshTokenFactory(user=user, application=application, access_token=access_token)
def assert_access_token_destroyed(self, user):
"""Assert all access tokens are destroyed."""
self.assertFalse(dot_access_token.objects.filter(user=user).exists())
self.assertFalse(dot_refresh_token.objects.filter(user=user).exists())
self.assertFalse(dop_access_token.objects.filter(user=user).exists())
self.assertFalse(dop_refresh_token.objects.filter(user=user).exists())
@ddt.ddt
class StudentAccountLoginAndRegistrationTest(ThirdPartyAuthTestMixin, UrlResetMixin, ModuleStoreTestCase):
""" Tests for the student account views that update the user's account information. """
shard = 7
USERNAME = "bob"
EMAIL = "bob@example.com"
PASSWORD = "password"
URLCONF_MODULES = ['openedx.core.djangoapps.embargo']
@mock.patch.dict(settings.FEATURES, {'EMBARGO': True})
def setUp(self):
super(StudentAccountLoginAndRegistrationTest, self).setUp()
# Several third party auth providers are created for these tests:
self.google_provider = self.configure_google_provider(enabled=True, visible=True)
self.configure_facebook_provider(enabled=True, visible=True)
self.configure_dummy_provider(
visible=True,
enabled=True,
icon_class='',
icon_image=SimpleUploadedFile('icon.svg', '<svg><rect width="50" height="100"/></svg>'),
)
self.hidden_enabled_provider = self.configure_linkedin_provider(
visible=False,
enabled=True,
)
self.hidden_disabled_provider = self.configure_azure_ad_provider()
@ddt.data(
("signin_user", "login"),
("register_user", "register"),
)
@ddt.unpack
def test_login_and_registration_form(self, url_name, initial_mode):
response = self.client.get(reverse(url_name))
expected_data = '"initial_mode": "{mode}"'.format(mode=initial_mode)
self.assertContains(response, expected_data)
@ddt.data("signin_user", "register_user")
def test_login_and_registration_form_already_authenticated(self, url_name):
# Create/activate a new account and log in
activation_key = create_account(self.USERNAME, self.PASSWORD, self.EMAIL)
activate_account(activation_key)
result = self.client.login(username=self.USERNAME, password=self.PASSWORD)
self.assertTrue(result)
# Verify that we're redirected to the dashboard
response = self.client.get(reverse(url_name))
self.assertRedirects(response, reverse("dashboard"))
@ddt.data(
(None, "signin_user"),
(None, "register_user"),
("edx.org", "signin_user"),
("edx.org", "register_user"),
)
@ddt.unpack
def test_login_and_registration_form_signin_not_preserves_params(self, theme, url_name):
params = [
('course_id', 'edX/DemoX/Demo_Course'),
('enrollment_action', 'enroll'),
]
# The response should not have a "Sign In" button with the URL
# that preserves the querystring params
with with_comprehensive_theme_context(theme):
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
expected_url = '/login?{}'.format(self._finish_auth_url_param(params + [('next', '/dashboard')]))
self.assertNotContains(response, expected_url)
# Add additional parameters:
params = [
('course_id', 'edX/DemoX/Demo_Course'),
('enrollment_action', 'enroll'),
('course_mode', CourseMode.DEFAULT_MODE_SLUG),
('email_opt_in', 'true'),
('next', '/custom/final/destination')
]
# Verify that this parameter is also preserved
with with_comprehensive_theme_context(theme):
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
expected_url = '/login?{}'.format(self._finish_auth_url_param(params))
self.assertNotContains(response, expected_url)
@mock.patch.dict(settings.FEATURES, {"ENABLE_THIRD_PARTY_AUTH": False})
@ddt.data("signin_user", "register_user")
def test_third_party_auth_disabled(self, url_name):
response = self.client.get(reverse(url_name))
self._assert_third_party_auth_data(response, None, None, [], None)
@mock.patch('student_account.views.enterprise_customer_for_request')
@mock.patch('openedx.core.djangoapps.user_api.api.enterprise_customer_for_request')
@ddt.data(
("signin_user", None, None, None, False),
("register_user", None, None, None, False),
("signin_user", "google-oauth2", "Google", None, False),
("register_user", "google-oauth2", "Google", None, False),
("signin_user", "facebook", "Facebook", None, False),
("register_user", "facebook", "Facebook", None, False),
("signin_user", "dummy", "Dummy", None, False),
("register_user", "dummy", "Dummy", None, False),
(
"signin_user",
"google-oauth2",
"Google",
{
'name': 'FakeName',
'logo': 'https://host.com/logo.jpg',
'welcome_msg': 'No message'
},
True
)
)
@ddt.unpack
def test_third_party_auth(
self,
url_name,
current_backend,
current_provider,
expected_enterprise_customer_mock_attrs,
add_user_details,
enterprise_customer_mock_1,
enterprise_customer_mock_2
):
params = [
('course_id', 'course-v1:Org+Course+Run'),
('enrollment_action', 'enroll'),
('course_mode', CourseMode.DEFAULT_MODE_SLUG),
('email_opt_in', 'true'),
('next', '/custom/final/destination'),
]
if expected_enterprise_customer_mock_attrs:
expected_ec = {
'name': expected_enterprise_customer_mock_attrs['name'],
'branding_configuration': {
'logo': 'https://host.com/logo.jpg',
'welcome_message': expected_enterprise_customer_mock_attrs['welcome_msg']
}
}
else:
expected_ec = None
email = None
if add_user_details:
email = 'test@test.com'
enterprise_customer_mock_1.return_value = expected_ec
enterprise_customer_mock_2.return_value = expected_ec
# Simulate a running pipeline
if current_backend is not None:
pipeline_target = "student_account.views.third_party_auth.pipeline"
with simulate_running_pipeline(pipeline_target, current_backend, email=email):
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
# Do NOT simulate a running pipeline
else:
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
# This relies on the THIRD_PARTY_AUTH configuration in the test settings
expected_providers = [
{
"id": "oa2-dummy",
"name": "Dummy",
"iconClass": None,
"iconImage": settings.MEDIA_URL + "icon.svg",
"loginUrl": self._third_party_login_url("dummy", "login", params),
"registerUrl": self._third_party_login_url("dummy", "register", params)
},
{
"id": "oa2-facebook",
"name": "Facebook",
"iconClass": "fa-facebook",
"iconImage": None,
"loginUrl": self._third_party_login_url("facebook", "login", params),
"registerUrl": self._third_party_login_url("facebook", "register", params)
},
{
"id": "oa2-google-oauth2",
"name": "Google",
"iconClass": "fa-google-plus",
"iconImage": None,
"loginUrl": self._third_party_login_url("google-oauth2", "login", params),
"registerUrl": self._third_party_login_url("google-oauth2", "register", params)
},
]
self._assert_third_party_auth_data(
response,
current_backend,
current_provider,
expected_providers,
expected_ec,
add_user_details
)
def _configure_testshib_provider(self, provider_name, idp_slug):
"""
Enable and configure the TestShib SAML IdP as a third_party_auth provider.
"""
kwargs = {}
kwargs.setdefault('name', provider_name)
kwargs.setdefault('enabled', True)
kwargs.setdefault('visible', True)
kwargs.setdefault('slug', idp_slug)
kwargs.setdefault('entity_id', 'https://idp.testshib.org/idp/shibboleth')
kwargs.setdefault('metadata_source', 'https://mock.testshib.org/metadata/testshib-providers.xml')
kwargs.setdefault('icon_class', 'fa-university')
kwargs.setdefault('attr_email', 'dummy-email-attr')
kwargs.setdefault('max_session_length', None)
self.configure_saml_provider(**kwargs)
@mock.patch('django.conf.settings.MESSAGE_STORAGE', 'django.contrib.messages.storage.cookie.CookieStorage')
@mock.patch('lms.djangoapps.student_account.views.enterprise_customer_for_request')
@ddt.data(
(
'signin_user',
'tpa-saml',
'TestShib',
)
)
@ddt.unpack
def test_saml_auth_with_error(
self,
url_name,
current_backend,
current_provider,
enterprise_customer_mock,
):
params = []
request = RequestFactory().get(reverse(url_name), params, HTTP_ACCEPT='text/html')
SessionMiddleware().process_request(request)
request.user = AnonymousUser()
self.enable_saml()
dummy_idp = 'testshib'
self._configure_testshib_provider(current_provider, dummy_idp)
enterprise_customer_data = {
'uuid': '72416e52-8c77-4860-9584-15e5b06220fb',
'name': 'Dummy Enterprise',
'identity_provider': dummy_idp,
}
enterprise_customer_mock.return_value = enterprise_customer_data
dummy_error_message = 'Authentication failed: SAML login failed ' \
'["invalid_response"] [SAML Response must contain 1 assertion]'
# Add error message for error in auth pipeline
MessageMiddleware().process_request(request)
messages.error(request, dummy_error_message, extra_tags='social-auth')
# Simulate a running pipeline
pipeline_response = {
'response': {
'idp_name': dummy_idp
}
}
pipeline_target = 'student_account.views.third_party_auth.pipeline'
with simulate_running_pipeline(pipeline_target, current_backend, **pipeline_response):
with mock.patch('edxmako.request_context.get_current_request', return_value=request):
response = login_and_registration_form(request)
expected_error_message = Text(_(
u'We are sorry, you are not authorized to access {platform_name} via this channel. '
u'Please contact your learning administrator or manager in order to access {platform_name}.'
u'{line_break}{line_break}'
u'Error Details:{line_break}{error_message}')
).format(
platform_name=settings.PLATFORM_NAME,
error_message=dummy_error_message,
line_break=HTML('<br/>')
)
self._assert_saml_auth_data_with_error(
response,
current_backend,
current_provider,
expected_error_message
)
def test_hinted_login(self):
params = [("next", "/courses/something/?tpa_hint=oa2-google-oauth2")]
response = self.client.get(reverse('signin_user'), params, HTTP_ACCEPT="text/html")
self.assertContains(response, '"third_party_auth_hint": "oa2-google-oauth2"')
tpa_hint = self.hidden_enabled_provider.provider_id
params = [("next", "/courses/something/?tpa_hint={0}".format(tpa_hint))]
response = self.client.get(reverse('signin_user'), params, HTTP_ACCEPT="text/html")
self.assertContains(response, '"third_party_auth_hint": "{0}"'.format(tpa_hint))
tpa_hint = self.hidden_disabled_provider.provider_id
params = [("next", "/courses/something/?tpa_hint={0}".format(tpa_hint))]
response = self.client.get(reverse('signin_user'), params, HTTP_ACCEPT="text/html")
self.assertNotIn(response.content, tpa_hint)
@ddt.data(
('signin_user', 'login'),
('register_user', 'register'),
)
@ddt.unpack
def test_hinted_login_dialog_disabled(self, url_name, auth_entry):
"""Test that the dialog doesn't show up for hinted logins when disabled. """
self.google_provider.skip_hinted_login_dialog = True
self.google_provider.save()
params = [("next", "/courses/something/?tpa_hint=oa2-google-oauth2")]
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
expected_url = '/auth/login/google-oauth2/?auth_entry={}&next=%2Fcourses'\
'%2Fsomething%2F%3Ftpa_hint%3Doa2-google-oauth2'.format(auth_entry)
self.assertRedirects(
response,
expected_url,
target_status_code=302
)
@override_settings(FEATURES=dict(settings.FEATURES, THIRD_PARTY_AUTH_HINT='oa2-google-oauth2'))
@ddt.data(
'signin_user',
'register_user',
)
def test_settings_tpa_hinted_login(self, url_name):
"""
Ensure that settings.FEATURES['THIRD_PARTY_AUTH_HINT'] can set third_party_auth_hint.
"""
params = [("next", "/courses/something/")]
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
self.assertContains(response, '"third_party_auth_hint": "oa2-google-oauth2"')
# THIRD_PARTY_AUTH_HINT can be overridden via the query string
tpa_hint = self.hidden_enabled_provider.provider_id
params = [("next", "/courses/something/?tpa_hint={0}".format(tpa_hint))]
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
self.assertContains(response, '"third_party_auth_hint": "{0}"'.format(tpa_hint))
# Even disabled providers in the query string will override THIRD_PARTY_AUTH_HINT
tpa_hint = self.hidden_disabled_provider.provider_id
params = [("next", "/courses/something/?tpa_hint={0}".format(tpa_hint))]
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
self.assertNotIn(response.content, tpa_hint)
@override_settings(FEATURES=dict(settings.FEATURES, THIRD_PARTY_AUTH_HINT='oa2-google-oauth2'))
@ddt.data(
('signin_user', 'login'),
('register_user', 'register'),
)
@ddt.unpack
def test_settings_tpa_hinted_login_dialog_disabled(self, url_name, auth_entry):
"""Test that the dialog doesn't show up for hinted logins when disabled via settings.THIRD_PARTY_AUTH_HINT. """
self.google_provider.skip_hinted_login_dialog = True
self.google_provider.save()
params = [("next", "/courses/something/")]
response = self.client.get(reverse(url_name), params, HTTP_ACCEPT="text/html")
expected_url = '/auth/login/google-oauth2/?auth_entry={}&next=%2Fcourses'\
'%2Fsomething%2F%3Ftpa_hint%3Doa2-google-oauth2'.format(auth_entry)
self.assertRedirects(
response,
expected_url,
target_status_code=302
)
@mock.patch('student_account.views.enterprise_customer_for_request')
@ddt.data(
('signin_user', False, None, None),
('register_user', False, None, None),
('signin_user', True, 'Fake EC', 'http://logo.com/logo.jpg'),
('register_user', True, 'Fake EC', 'http://logo.com/logo.jpg'),
('signin_user', True, 'Fake EC', None),
('register_user', True, 'Fake EC', None),
)
@ddt.unpack
def test_enterprise_register(self, url_name, ec_present, ec_name, logo_url, mock_get_ec):
"""
Verify that when an EnterpriseCustomer is received on the login and register views,
the appropriate sidebar is rendered.
"""
if ec_present:
mock_get_ec.return_value = {
'name': ec_name,
'branding_configuration': {'logo': logo_url}
}
else:
mock_get_ec.return_value = None
response = self.client.get(reverse(url_name), HTTP_ACCEPT="text/html")
enterprise_sidebar_div_id = u'enterprise-content-container'
if not ec_present:
self.assertNotContains(response, text=enterprise_sidebar_div_id)
else:
self.assertContains(response, text=enterprise_sidebar_div_id)
welcome_message = settings.ENTERPRISE_SPECIFIC_BRANDED_WELCOME_TEMPLATE
expected_message = Text(welcome_message).format(
start_bold=HTML('<b>'),
end_bold=HTML('</b>'),
line_break=HTML('<br/>'),
enterprise_name=ec_name,
platform_name=settings.PLATFORM_NAME,
privacy_policy_link_start=HTML("<a href='{pp_url}' target='_blank'>").format(
pp_url=settings.MKTG_URLS.get('PRIVACY', 'https://www.edx.org/edx-privacy-policy')
),
privacy_policy_link_end=HTML("</a>"),
)
self.assertContains(response, expected_message)
if logo_url:
self.assertContains(response, logo_url)
def test_enterprise_cookie_delete(self):
"""
Test that enterprise cookies are deleted in login/registration views.
Cookies must be deleted in login/registration views so that *default* login/registration branding
is displayed to subsequent requests from non-enterprise customers.
"""
cookies = SimpleCookie()
cookies[settings.ENTERPRISE_CUSTOMER_COOKIE_NAME] = 'test-enterprise-customer'
response = self.client.get(reverse('signin_user'), HTTP_ACCEPT="text/html", cookies=cookies)
self.assertIn(settings.ENTERPRISE_CUSTOMER_COOKIE_NAME, response.cookies)
enterprise_cookie = response.cookies[settings.ENTERPRISE_CUSTOMER_COOKIE_NAME]
self.assertEqual(enterprise_cookie['domain'], settings.BASE_COOKIE_DOMAIN)
self.assertEqual(enterprise_cookie.value, '')
@override_settings(SITE_NAME=settings.MICROSITE_TEST_HOSTNAME)
def test_microsite_uses_old_login_page(self):
# Retrieve the login page from a microsite domain
# and verify that we're served the old page.
resp = self.client.get(
reverse("signin_user"),
HTTP_HOST=settings.MICROSITE_TEST_HOSTNAME
)
self.assertContains(resp, "Log into your Test Site Account")
self.assertContains(resp, "login-form")
def test_microsite_uses_old_register_page(self):
# Retrieve the register page from a microsite domain
# and verify that we're served the old page.
resp = self.client.get(
reverse("register_user"),
HTTP_HOST=settings.MICROSITE_TEST_HOSTNAME
)
self.assertContains(resp, "Register for Test Site")
self.assertContains(resp, "register-form")
def test_login_registration_xframe_protected(self):
resp = self.client.get(
reverse("register_user"),
{},
HTTP_REFERER="http://localhost/iframe"
)
self.assertEqual(resp['X-Frame-Options'], 'DENY')
self.configure_lti_provider(name='Test', lti_hostname='localhost', lti_consumer_key='test_key', enabled=True)
resp = self.client.get(
reverse("register_user"),
HTTP_REFERER="http://localhost/iframe"
)
self.assertEqual(resp['X-Frame-Options'], 'ALLOW')
def _assert_third_party_auth_data(self, response, current_backend, current_provider, providers, expected_ec,
add_user_details=False):
"""Verify that third party auth info is rendered correctly in a DOM data attribute. """
finish_auth_url = None
if current_backend:
finish_auth_url = reverse("social:complete", kwargs={"backend": current_backend}) + "?"
auth_info = {
"currentProvider": current_provider,
"providers": providers,
"secondaryProviders": [],
"finishAuthUrl": finish_auth_url,
"errorMessage": None,
"registerFormSubmitButtonText": "Create Account",
"syncLearnerProfileData": False,
"pipeline_user_details": {"email": "test@test.com"} if add_user_details else {}
}
if expected_ec is not None:
# If we set an EnterpriseCustomer, third-party auth providers ought to be hidden.
auth_info['providers'] = []
auth_info = dump_js_escaped_json(auth_info)
expected_data = '"third_party_auth": {auth_info}'.format(
auth_info=auth_info
)
self.assertContains(response, expected_data)
def _assert_saml_auth_data_with_error(
self, response, current_backend, current_provider, expected_error_message
):
"""
Verify that third party auth info is rendered correctly in a DOM data attribute.
"""
finish_auth_url = None
if current_backend:
finish_auth_url = reverse('social:complete', kwargs={'backend': current_backend}) + '?'
auth_info = {
'currentProvider': current_provider,
'providers': [],
'secondaryProviders': [],
'finishAuthUrl': finish_auth_url,
'errorMessage': expected_error_message,
'registerFormSubmitButtonText': 'Create Account',
'syncLearnerProfileData': False,
'pipeline_user_details': {'response': {'idp_name': 'testshib'}}
}
auth_info = dump_js_escaped_json(auth_info)
expected_data = '"third_party_auth": {auth_info}'.format(
auth_info=auth_info
)
self.assertContains(response, expected_data)
def _third_party_login_url(self, backend_name, auth_entry, login_params):
"""Construct the login URL to start third party authentication. """
return u"{url}?auth_entry={auth_entry}&{param_str}".format(
url=reverse("social:begin", kwargs={"backend": backend_name}),
auth_entry=auth_entry,
param_str=self._finish_auth_url_param(login_params),
)
def _finish_auth_url_param(self, params):
"""
Make the next=... URL parameter that indicates where the user should go next.
>>> _finish_auth_url_param([('next', '/dashboard')])
'/account/finish_auth?next=%2Fdashboard'
"""
return urlencode({
'next': '/account/finish_auth?{}'.format(urlencode(params))
})
def test_english_by_default(self):
response = self.client.get(reverse('signin_user'), [], HTTP_ACCEPT="text/html")
self.assertEqual(response['Content-Language'], 'en')
def test_unsupported_language(self):
response = self.client.get(reverse('signin_user'), [], HTTP_ACCEPT="text/html", HTTP_ACCEPT_LANGUAGE="ts-zx")
self.assertEqual(response['Content-Language'], 'en')
def test_browser_language(self):
response = self.client.get(reverse('signin_user'), [], HTTP_ACCEPT="text/html", HTTP_ACCEPT_LANGUAGE="es")
self.assertEqual(response['Content-Language'], 'es-419')
def test_browser_language_dialent(self):
response = self.client.get(reverse('signin_user'), [], HTTP_ACCEPT="text/html", HTTP_ACCEPT_LANGUAGE="es-es")
self.assertEqual(response['Content-Language'], 'es-es')
class AccountSettingsViewTest(ThirdPartyAuthTestMixin, TestCase, ProgramsApiConfigMixin):
""" Tests for the account settings view. """
USERNAME = 'student'
PASSWORD = 'password'
FIELDS = [
'country',
'gender',
'language',
'level_of_education',
'password',
'year_of_birth',
'preferred_language',
'time_zone',
]
@mock.patch("django.conf.settings.MESSAGE_STORAGE", 'django.contrib.messages.storage.cookie.CookieStorage')
def setUp(self):
super(AccountSettingsViewTest, self).setUp()
self.user = UserFactory.create(username=self.USERNAME, password=self.PASSWORD)
CommerceConfiguration.objects.create(cache_ttl=10, enabled=True)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
self.request = HttpRequest()
self.request.user = self.user
# For these tests, two third party auth providers are enabled by default:
self.configure_google_provider(enabled=True, visible=True)
self.configure_facebook_provider(enabled=True, visible=True)
# Python-social saves auth failure notifcations in Django messages.
# See pipeline.get_duplicate_provider() for details.
self.request.COOKIES = {}
MessageMiddleware().process_request(self.request)
messages.error(self.request, 'Facebook is already in use.', extra_tags='Auth facebook')
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_customer_for_learner')
def test_context(self, mock_get_enterprise_customer_for_learner):
self.request.site = SiteFactory.create()
mock_get_enterprise_customer_for_learner.return_value = {}
context = account_settings_context(self.request)
user_accounts_api_url = reverse("accounts_api", kwargs={'username': self.user.username})
self.assertEqual(context['user_accounts_api_url'], user_accounts_api_url)
user_preferences_api_url = reverse('preferences_api', kwargs={'username': self.user.username})
self.assertEqual(context['user_preferences_api_url'], user_preferences_api_url)
for attribute in self.FIELDS:
self.assertIn(attribute, context['fields'])
self.assertEqual(
context['user_accounts_api_url'], reverse("accounts_api", kwargs={'username': self.user.username})
)
self.assertEqual(
context['user_preferences_api_url'], reverse('preferences_api', kwargs={'username': self.user.username})
)
self.assertEqual(context['duplicate_provider'], 'facebook')
self.assertEqual(context['auth']['providers'][0]['name'], 'Facebook')
self.assertEqual(context['auth']['providers'][1]['name'], 'Google')
self.assertEqual(context['sync_learner_profile_data'], False)
self.assertEqual(context['edx_support_url'], settings.SUPPORT_SITE_LINK)
self.assertEqual(context['enterprise_name'], None)
self.assertEqual(
context['enterprise_readonly_account_fields'], {'fields': settings.ENTERPRISE_READONLY_ACCOUNT_FIELDS}
)
@mock.patch('student_account.views.get_enterprise_customer_for_learner')
@mock.patch('openedx.features.enterprise_support.utils.third_party_auth.provider.Registry.get')
def test_context_for_enterprise_learner(
self, mock_get_auth_provider, mock_get_enterprise_customer_for_learner
):
dummy_enterprise_customer = {
'uuid': 'real-ent-uuid',
'name': 'Dummy Enterprise',
'identity_provider': 'saml-ubc'
}
mock_get_enterprise_customer_for_learner.return_value = dummy_enterprise_customer
self.request.site = SiteFactory.create()
mock_get_auth_provider.return_value.sync_learner_profile_data = True
context = account_settings_context(self.request)
user_accounts_api_url = reverse("accounts_api", kwargs={'username': self.user.username})
self.assertEqual(context['user_accounts_api_url'], user_accounts_api_url)
user_preferences_api_url = reverse('preferences_api', kwargs={'username': self.user.username})
self.assertEqual(context['user_preferences_api_url'], user_preferences_api_url)
for attribute in self.FIELDS:
self.assertIn(attribute, context['fields'])
self.assertEqual(
context['user_accounts_api_url'], reverse("accounts_api", kwargs={'username': self.user.username})
)
self.assertEqual(
context['user_preferences_api_url'], reverse('preferences_api', kwargs={'username': self.user.username})
)
self.assertEqual(context['duplicate_provider'], 'facebook')
self.assertEqual(context['auth']['providers'][0]['name'], 'Facebook')
self.assertEqual(context['auth']['providers'][1]['name'], 'Google')
self.assertEqual(
context['sync_learner_profile_data'], mock_get_auth_provider.return_value.sync_learner_profile_data
)
self.assertEqual(context['edx_support_url'], settings.SUPPORT_SITE_LINK)
self.assertEqual(context['enterprise_name'], dummy_enterprise_customer['name'])
self.assertEqual(
context['enterprise_readonly_account_fields'], {'fields': settings.ENTERPRISE_READONLY_ACCOUNT_FIELDS}
)
def test_view(self):
"""
Test that all fields are visible
"""
view_path = reverse('account_settings')
response = self.client.get(path=view_path)
for attribute in self.FIELDS:
self.assertIn(attribute, response.content)
def test_header_with_programs_listing_enabled(self):
"""
Verify that tabs header will be shown while program listing is enabled.
"""
self.create_programs_config()
view_path = reverse('account_settings')
response = self.client.get(path=view_path)
self.assertContains(response, 'global-header')
def test_header_with_programs_listing_disabled(self):
"""
Verify that nav header will be shown while program listing is disabled.
"""
self.create_programs_config(enabled=False)
view_path = reverse('account_settings')
response = self.client.get(path=view_path)
self.assertContains(response, 'global-header')
def test_commerce_order_detail(self):
"""
Verify that get_user_orders returns the correct order data.
"""
with mock_get_orders():
order_detail = get_user_orders(self.user)
for i, order in enumerate(mock_get_orders.default_response['results']):
expected = {
'number': order['number'],
'price': order['total_excl_tax'],
'order_date': 'Jan 01, 2016',
'receipt_url': '/checkout/receipt/?order_number=' + order['number'],
'lines': order['lines'],
}
self.assertEqual(order_detail[i], expected)
def test_commerce_order_detail_exception(self):
with mock_get_orders(exception=exceptions.HttpNotFoundError):
order_detail = get_user_orders(self.user)
self.assertEqual(order_detail, [])
def test_incomplete_order_detail(self):
response = {
'results': [
factories.OrderFactory(
status='Incomplete',
lines=[
factories.OrderLineFactory(
product=factories.ProductFactory(attribute_values=[factories.ProductAttributeFactory()])
)
]
)
]
}
with mock_get_orders(response=response):
order_detail = get_user_orders(self.user)
self.assertEqual(order_detail, [])
def test_order_history_with_no_product(self):
response = {
'results': [
factories.OrderFactory(
lines=[
factories.OrderLineFactory(
product=None
),
factories.OrderLineFactory(
product=factories.ProductFactory(attribute_values=[factories.ProductAttributeFactory(
name='certificate_type',
value='verified'
)])
)
]
)
]
}
with mock_get_orders(response=response):
order_detail = get_user_orders(self.user)
self.assertEqual(len(order_detail), 1)
@override_settings(SITE_NAME=settings.MICROSITE_LOGISTRATION_HOSTNAME)
class MicrositeLogistrationTests(TestCase):
"""
Test to validate that microsites can display the logistration page
"""
def test_login_page(self):
"""
Make sure that we get the expected logistration page on our specialized
microsite
"""
resp = self.client.get(
reverse('signin_user'),
HTTP_HOST=settings.MICROSITE_LOGISTRATION_HOSTNAME
)
self.assertEqual(resp.status_code, 200)
self.assertIn('<div id="login-and-registration-container"', resp.content)
def test_registration_page(self):
"""
Make sure that we get the expected logistration page on our specialized
microsite
"""
resp = self.client.get(
reverse('register_user'),
HTTP_HOST=settings.MICROSITE_LOGISTRATION_HOSTNAME
)
self.assertEqual(resp.status_code, 200)
self.assertIn('<div id="login-and-registration-container"', resp.content)
@override_settings(SITE_NAME=settings.MICROSITE_TEST_HOSTNAME)
def test_no_override(self):
"""
Make sure we get the old style login/registration if we don't override
"""
resp = self.client.get(
reverse('signin_user'),
HTTP_HOST=settings.MICROSITE_TEST_HOSTNAME
)
self.assertEqual(resp.status_code, 200)
self.assertNotIn('<div id="login-and-registration-container"', resp.content)
resp = self.client.get(
reverse('register_user'),
HTTP_HOST=settings.MICROSITE_TEST_HOSTNAME
)
self.assertEqual(resp.status_code, 200)
self.assertNotIn('<div id="login-and-registration-container"', resp.content)
class AccountCreationTestCaseWithSiteOverrides(SiteMixin, TestCase):
"""
Test cases for Feature flag ALLOW_PUBLIC_ACCOUNT_CREATION which when
turned off disables the account creation options in lms
"""
def setUp(self):
"""Set up the tests"""
super(AccountCreationTestCaseWithSiteOverrides, self).setUp()
# Set the feature flag ALLOW_PUBLIC_ACCOUNT_CREATION to False
self.site_configuration_values = {
'ALLOW_PUBLIC_ACCOUNT_CREATION': False
}
self.site_domain = 'testserver1.com'
self.set_up_site(self.site_domain, self.site_configuration_values)
def test_register_option_login_page(self):
"""
Navigate to the login page and check the Register option is hidden when
ALLOW_PUBLIC_ACCOUNT_CREATION flag is turned off
"""
response = self.client.get(reverse('signin_user'))
self.assertNotIn('<a class="btn-neutral" href="/register?next=%2Fdashboard">Register</a>',
response.content)

View File

@@ -1,14 +0,0 @@
from django.conf import settings
from django.conf.urls import url
from student_account import views
urlpatterns = [
url(r'^finish_auth$', views.finish_auth, name='finish_auth'),
url(r'^settings$', views.account_settings, name='account_settings'),
]
if settings.FEATURES.get('ENABLE_COMBINED_LOGIN_REGISTRATION'):
urlpatterns += [
url(r'^password$', views.password_change_request_handler, name='password_change_request'),
]

View File

@@ -1,606 +0,0 @@
""" Views for a student's account information. """
import json
import logging
from datetime import datetime
import urlparse
from django.conf import settings
from django.contrib import messages
from django.contrib.auth import get_user_model
from django.contrib.auth.decorators import login_required
from django.urls import reverse
from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest, HttpResponseForbidden
from django.shortcuts import redirect
from django.utils.translation import ugettext as _
from django.views.decorators.csrf import ensure_csrf_cookie
from django.views.decorators.http import require_http_methods
from django_countries import countries
import third_party_auth
from edx_ace import ace
from edx_ace.recipient import Recipient
from edxmako.shortcuts import render_to_response
from lms.djangoapps.commerce.models import CommerceConfiguration
from lms.djangoapps.commerce.utils import EcommerceService
from openedx.core.djangoapps.ace_common.template_context import get_base_template_context
from openedx.core.djangoapps.commerce.utils import ecommerce_api_client
from openedx.core.djangoapps.external_auth.login_and_register import login as external_auth_login
from openedx.core.djangoapps.external_auth.login_and_register import register as external_auth_register
from openedx.core.djangoapps.lang_pref.api import all_languages, released_languages
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
from openedx.core.djangoapps.theming.helpers import is_request_in_themed_site, get_current_site
from openedx.core.djangoapps.user_api.accounts.api import request_password_change
from openedx.core.djangoapps.user_api.api import (
RegistrationFormFactory,
get_login_session_form,
get_password_reset_form
)
from openedx.core.djangoapps.user_api.errors import (
UserNotFound,
UserAPIInternalError
)
from openedx.core.lib.edx_api_utils import get_edx_api_data
from openedx.core.lib.time_zone_utils import TIME_ZONE_CHOICES
from openedx.features.enterprise_support.api import enterprise_customer_for_request, get_enterprise_customer_for_learner
from openedx.features.enterprise_support.utils import (
handle_enterprise_cookies_for_logistration,
update_logistration_context_for_enterprise,
update_account_settings_context_for_enterprise,
)
from student.helpers import destroy_oauth_tokens, get_next_url_for_login_page
from student.message_types import PasswordReset
from student.models import UserProfile
from student.views import register_user as old_register_view, signin_user as old_login_view
from third_party_auth import pipeline
from third_party_auth.decorators import xframe_allow_whitelisted
from util.bad_request_rate_limiter import BadRequestRateLimiter
from util.date_utils import strftime_localized
AUDIT_LOG = logging.getLogger("audit")
log = logging.getLogger(__name__)
User = get_user_model() # pylint:disable=invalid-name
@require_http_methods(['GET'])
@ensure_csrf_cookie
@xframe_allow_whitelisted
def login_and_registration_form(request, initial_mode="login"):
"""Render the combined login/registration form, defaulting to login
This relies on the JS to asynchronously load the actual form from
the user_api.
Keyword Args:
initial_mode (string): Either "login" or "register".
"""
# Determine the URL to redirect to following login/registration/third_party_auth
redirect_to = get_next_url_for_login_page(request)
# If we're already logged in, redirect to the dashboard
if request.user.is_authenticated:
return redirect(redirect_to)
# Retrieve the form descriptions from the user API
form_descriptions = _get_form_descriptions(request)
# Our ?next= URL may itself contain a parameter 'tpa_hint=x' that we need to check.
# If present, we display a login page focused on third-party auth with that provider.
third_party_auth_hint = None
if '?' in redirect_to:
try:
next_args = urlparse.parse_qs(urlparse.urlparse(redirect_to).query)
provider_id = next_args['tpa_hint'][0]
tpa_hint_provider = third_party_auth.provider.Registry.get(provider_id=provider_id)
if tpa_hint_provider:
if tpa_hint_provider.skip_hinted_login_dialog:
# Forward the user directly to the provider's login URL when the provider is configured
# to skip the dialog.
if initial_mode == "register":
auth_entry = pipeline.AUTH_ENTRY_REGISTER
else:
auth_entry = pipeline.AUTH_ENTRY_LOGIN
return redirect(
pipeline.get_login_url(provider_id, auth_entry, redirect_url=redirect_to)
)
third_party_auth_hint = provider_id
initial_mode = "hinted_login"
except (KeyError, ValueError, IndexError) as ex:
log.error("Unknown tpa_hint provider: %s", ex)
# If this is a themed site, revert to the old login/registration pages.
# We need to do this for now to support existing themes.
# Themed sites can use the new logistration page by setting
# 'ENABLE_COMBINED_LOGIN_REGISTRATION' in their
# configuration settings.
if is_request_in_themed_site() and not configuration_helpers.get_value('ENABLE_COMBINED_LOGIN_REGISTRATION', False):
if initial_mode == "login":
return old_login_view(request)
elif initial_mode == "register":
return old_register_view(request)
# Allow external auth to intercept and handle the request
ext_auth_response = _external_auth_intercept(request, initial_mode)
if ext_auth_response is not None:
return ext_auth_response
# Account activation message
account_activation_messages = [
{
'message': message.message, 'tags': message.tags
} for message in messages.get_messages(request) if 'account-activation' in message.tags
]
# Otherwise, render the combined login/registration page
context = {
'data': {
'login_redirect_url': redirect_to,
'initial_mode': initial_mode,
'third_party_auth': _third_party_auth_context(request, redirect_to, third_party_auth_hint),
'third_party_auth_hint': third_party_auth_hint or '',
'platform_name': configuration_helpers.get_value('PLATFORM_NAME', settings.PLATFORM_NAME),
'support_link': configuration_helpers.get_value('SUPPORT_SITE_LINK', settings.SUPPORT_SITE_LINK),
'password_reset_support_link': configuration_helpers.get_value(
'PASSWORD_RESET_SUPPORT_LINK', settings.PASSWORD_RESET_SUPPORT_LINK
) or settings.SUPPORT_SITE_LINK,
'account_activation_messages': account_activation_messages,
# Include form descriptions retrieved from the user API.
# We could have the JS client make these requests directly,
# but we include them in the initial page load to avoid
# the additional round-trip to the server.
'login_form_desc': json.loads(form_descriptions['login']),
'registration_form_desc': json.loads(form_descriptions['registration']),
'password_reset_form_desc': json.loads(form_descriptions['password_reset']),
'account_creation_allowed': configuration_helpers.get_value(
'ALLOW_PUBLIC_ACCOUNT_CREATION', settings.FEATURES.get('ALLOW_PUBLIC_ACCOUNT_CREATION', True))
},
'login_redirect_url': redirect_to, # This gets added to the query string of the "Sign In" button in header
'responsive': True,
'allow_iframing': True,
'disable_courseware_js': True,
'combined_login_and_register': True,
'disable_footer': not configuration_helpers.get_value(
'ENABLE_COMBINED_LOGIN_REGISTRATION_FOOTER',
settings.FEATURES['ENABLE_COMBINED_LOGIN_REGISTRATION_FOOTER']
),
}
enterprise_customer = enterprise_customer_for_request(request)
update_logistration_context_for_enterprise(request, context, enterprise_customer)
response = render_to_response('student_account/login_and_register.html', context)
handle_enterprise_cookies_for_logistration(request, response, context)
return response
@require_http_methods(['POST'])
def password_change_request_handler(request):
"""Handle password change requests originating from the account page.
Uses the Account API to email the user a link to the password reset page.
Note:
The next step in the password reset process (confirmation) is currently handled
by student.views.password_reset_confirm_wrapper, a custom wrapper around Django's
password reset confirmation view.
Args:
request (HttpRequest)
Returns:
HttpResponse: 200 if the email was sent successfully
HttpResponse: 400 if there is no 'email' POST parameter
HttpResponse: 403 if the client has been rate limited
HttpResponse: 405 if using an unsupported HTTP method
Example usage:
POST /account/password
"""
limiter = BadRequestRateLimiter()
if limiter.is_rate_limit_exceeded(request):
AUDIT_LOG.warning("Password reset rate limit exceeded")
return HttpResponseForbidden()
user = request.user
# Prefer logged-in user's email
email = user.email if user.is_authenticated else request.POST.get('email')
if email:
try:
request_password_change(email, request.is_secure())
user = user if user.is_authenticated else User.objects.get(email=email)
destroy_oauth_tokens(user)
except UserNotFound:
AUDIT_LOG.info("Invalid password reset attempt")
# Increment the rate limit counter
limiter.tick_bad_request_counter(request)
# If enabled, send an email saying that a password reset was attempted, but that there is
# no user associated with the email
if configuration_helpers.get_value('ENABLE_PASSWORD_RESET_FAILURE_EMAIL',
settings.FEATURES['ENABLE_PASSWORD_RESET_FAILURE_EMAIL']):
site = get_current_site()
message_context = get_base_template_context(site)
message_context.update({
'failed': True,
'request': request, # Used by google_analytics_tracking_pixel
'email_address': email,
})
msg = PasswordReset().personalize(
recipient=Recipient(username='', email_address=email),
language=settings.LANGUAGE_CODE,
user_context=message_context,
)
ace.send(msg)
except UserAPIInternalError as err:
log.exception('Error occured during password change for user {email}: {error}'
.format(email=email, error=err))
return HttpResponse(_("Some error occured during password change. Please try again"), status=500)
return HttpResponse(status=200)
else:
return HttpResponseBadRequest(_("No email address provided."))
def _third_party_auth_context(request, redirect_to, tpa_hint=None):
"""Context for third party auth providers and the currently running pipeline.
Arguments:
request (HttpRequest): The request, used to determine if a pipeline
is currently running.
redirect_to: The URL to send the user to following successful
authentication.
tpa_hint (string): An override flag that will return a matching provider
as long as its configuration has been enabled
Returns:
dict
"""
context = {
"currentProvider": None,
"providers": [],
"secondaryProviders": [],
"finishAuthUrl": None,
"errorMessage": None,
"registerFormSubmitButtonText": _("Create Account"),
"syncLearnerProfileData": False,
"pipeline_user_details": {}
}
if third_party_auth.is_enabled():
for enabled in third_party_auth.provider.Registry.displayed_for_login(tpa_hint=tpa_hint):
info = {
"id": enabled.provider_id,
"name": enabled.name,
"iconClass": enabled.icon_class or None,
"iconImage": enabled.icon_image.url if enabled.icon_image else None,
"loginUrl": pipeline.get_login_url(
enabled.provider_id,
pipeline.AUTH_ENTRY_LOGIN,
redirect_url=redirect_to,
),
"registerUrl": pipeline.get_login_url(
enabled.provider_id,
pipeline.AUTH_ENTRY_REGISTER,
redirect_url=redirect_to,
),
}
context["providers" if not enabled.secondary else "secondaryProviders"].append(info)
running_pipeline = pipeline.get(request)
if running_pipeline is not None:
current_provider = third_party_auth.provider.Registry.get_from_pipeline(running_pipeline)
user_details = running_pipeline['kwargs']['details']
if user_details:
context['pipeline_user_details'] = user_details
if current_provider is not None:
context["currentProvider"] = current_provider.name
context["finishAuthUrl"] = pipeline.get_complete_url(current_provider.backend_name)
context["syncLearnerProfileData"] = current_provider.sync_learner_profile_data
if current_provider.skip_registration_form:
# As a reliable way of "skipping" the registration form, we just submit it automatically
context["autoSubmitRegForm"] = True
# Check for any error messages we may want to display:
for msg in messages.get_messages(request):
if msg.extra_tags.split()[0] == "social-auth":
# msg may or may not be translated. Try translating [again] in case we are able to:
context['errorMessage'] = _(unicode(msg))
break
return context
def _get_form_descriptions(request):
"""Retrieve form descriptions from the user API.
Arguments:
request (HttpRequest): The original request, used to retrieve session info.
Returns:
dict: Keys are 'login', 'registration', and 'password_reset';
values are the JSON-serialized form descriptions.
"""
return {
'password_reset': get_password_reset_form().to_json(),
'login': get_login_session_form(request).to_json(),
'registration': RegistrationFormFactory().get_registration_form(request).to_json()
}
def _get_extended_profile_fields():
"""Retrieve the extended profile fields from site configuration to be shown on the
Account Settings page
Returns:
A list of dicts. Each dict corresponds to a single field. The keys per field are:
"field_name" : name of the field stored in user_profile.meta
"field_label" : The label of the field.
"field_type" : TextField or ListField
"field_options": a list of tuples for options in the dropdown in case of ListField
"""
extended_profile_fields = []
fields_already_showing = ['username', 'name', 'email', 'pref-lang', 'country', 'time_zone', 'level_of_education',
'gender', 'year_of_birth', 'language_proficiencies', 'social_links']
field_labels_map = {
"first_name": _(u"First Name"),
"last_name": _(u"Last Name"),
"city": _(u"City"),
"state": _(u"State/Province/Region"),
"company": _(u"Company"),
"title": _(u"Title"),
"job_title": _(u"Job Title"),
"mailing_address": _(u"Mailing address"),
"goals": _(u"Tell us why you're interested in {platform_name}").format(
platform_name=configuration_helpers.get_value("PLATFORM_NAME", settings.PLATFORM_NAME)
),
"profession": _(u"Profession"),
"specialty": _(u"Specialty")
}
extended_profile_field_names = configuration_helpers.get_value('extended_profile_fields', [])
for field_to_exclude in fields_already_showing:
if field_to_exclude in extended_profile_field_names:
extended_profile_field_names.remove(field_to_exclude)
extended_profile_field_options = configuration_helpers.get_value('EXTRA_FIELD_OPTIONS', [])
extended_profile_field_option_tuples = {}
for field in extended_profile_field_options.keys():
field_options = extended_profile_field_options[field]
extended_profile_field_option_tuples[field] = [(option.lower(), option) for option in field_options]
for field in extended_profile_field_names:
field_dict = {
"field_name": field,
"field_label": field_labels_map.get(field, field),
}
field_options = extended_profile_field_option_tuples.get(field)
if field_options:
field_dict["field_type"] = "ListField"
field_dict["field_options"] = field_options
else:
field_dict["field_type"] = "TextField"
extended_profile_fields.append(field_dict)
return extended_profile_fields
def _external_auth_intercept(request, mode):
"""Allow external auth to intercept a login/registration request.
Arguments:
request (Request): The original request.
mode (str): Either "login" or "register"
Returns:
Response or None
"""
if mode == "login":
return external_auth_login(request)
elif mode == "register":
return external_auth_register(request)
def get_user_orders(user):
"""Given a user, get the detail of all the orders from the Ecommerce service.
Args:
user (User): The user to authenticate as when requesting ecommerce.
Returns:
list of dict, representing orders returned by the Ecommerce service.
"""
no_data = []
user_orders = []
commerce_configuration = CommerceConfiguration.current()
user_query = {'username': user.username}
use_cache = commerce_configuration.is_cache_enabled
cache_key = commerce_configuration.CACHE_KEY + '.' + str(user.id) if use_cache else None
api = ecommerce_api_client(user)
commerce_user_orders = get_edx_api_data(
commerce_configuration, 'orders', api=api, querystring=user_query, cache_key=cache_key
)
for order in commerce_user_orders:
if order['status'].lower() == 'complete':
date_placed = datetime.strptime(order['date_placed'], "%Y-%m-%dT%H:%M:%SZ")
order_data = {
'number': order['number'],
'price': order['total_excl_tax'],
'order_date': strftime_localized(date_placed, 'SHORT_DATE'),
'receipt_url': EcommerceService().get_receipt_page_url(order['number']),
'lines': order['lines'],
}
user_orders.append(order_data)
return user_orders
@login_required
@require_http_methods(['GET'])
def account_settings(request):
"""Render the current user's account settings page.
Args:
request (HttpRequest)
Returns:
HttpResponse: 200 if the page was sent successfully
HttpResponse: 302 if not logged in (redirect to login page)
HttpResponse: 405 if using an unsupported HTTP method
Example usage:
GET /account/settings
"""
context = account_settings_context(request)
return render_to_response('student_account/account_settings.html', context)
@login_required
@require_http_methods(['GET'])
def finish_auth(request): # pylint: disable=unused-argument
""" Following logistration (1st or 3rd party), handle any special query string params.
See FinishAuthView.js for details on the query string params.
e.g. auto-enroll the user in a course, set email opt-in preference.
This view just displays a "Please wait" message while AJAX calls are made to enroll the
user in the course etc. This view is only used if a parameter like "course_id" is present
during login/registration/third_party_auth. Otherwise, there is no need for it.
Ideally this view will finish and redirect to the next step before the user even sees it.
Args:
request (HttpRequest)
Returns:
HttpResponse: 200 if the page was sent successfully
HttpResponse: 302 if not logged in (redirect to login page)
HttpResponse: 405 if using an unsupported HTTP method
Example usage:
GET /account/finish_auth/?course_id=course-v1:blah&enrollment_action=enroll
"""
return render_to_response('student_account/finish_auth.html', {
'disable_courseware_js': True,
'disable_footer': True,
})
def account_settings_context(request):
""" Context for the account settings page.
Args:
request: The request object.
Returns:
dict
"""
user = request.user
year_of_birth_options = [(unicode(year), unicode(year)) for year in UserProfile.VALID_YEARS]
try:
user_orders = get_user_orders(user)
except: # pylint: disable=bare-except
log.exception('Error fetching order history from Otto.')
# Return empty order list as account settings page expect a list and
# it will be broken if exception raised
user_orders = []
context = {
'auth': {},
'duplicate_provider': None,
'nav_hidden': True,
'fields': {
'country': {
'options': list(countries),
}, 'gender': {
'options': [(choice[0], _(choice[1])) for choice in UserProfile.GENDER_CHOICES],
}, 'language': {
'options': released_languages(),
}, 'level_of_education': {
'options': [(choice[0], _(choice[1])) for choice in UserProfile.LEVEL_OF_EDUCATION_CHOICES],
}, 'password': {
'url': reverse('password_reset'),
}, 'year_of_birth': {
'options': year_of_birth_options,
}, 'preferred_language': {
'options': all_languages(),
}, 'time_zone': {
'options': TIME_ZONE_CHOICES,
}
},
'platform_name': configuration_helpers.get_value('PLATFORM_NAME', settings.PLATFORM_NAME),
'password_reset_support_link': configuration_helpers.get_value(
'PASSWORD_RESET_SUPPORT_LINK', settings.PASSWORD_RESET_SUPPORT_LINK
) or settings.SUPPORT_SITE_LINK,
'user_accounts_api_url': reverse("accounts_api", kwargs={'username': user.username}),
'user_preferences_api_url': reverse('preferences_api', kwargs={'username': user.username}),
'disable_courseware_js': True,
'show_program_listing': ProgramsApiConfig.is_enabled(),
'show_dashboard_tabs': True,
'order_history': user_orders,
'enable_account_deletion': configuration_helpers.get_value(
'ENABLE_ACCOUNT_DELETION', settings.FEATURES.get('ENABLE_ACCOUNT_DELETION', False)
),
'extended_profile_fields': _get_extended_profile_fields(),
}
enterprise_customer = get_enterprise_customer_for_learner(site=request.site, user=request.user)
update_account_settings_context_for_enterprise(context, enterprise_customer)
if third_party_auth.is_enabled():
# If the account on the third party provider is already connected with another edX account,
# we display a message to the user.
context['duplicate_provider'] = pipeline.get_duplicate_provider(messages.get_messages(request))
auth_states = pipeline.get_provider_user_states(user)
context['auth']['providers'] = [{
'id': state.provider.provider_id,
'name': state.provider.name, # The name of the provider e.g. Facebook
'connected': state.has_account, # Whether the user's edX account is connected with the provider.
# If the user is not connected, they should be directed to this page to authenticate
# with the particular provider, as long as the provider supports initiating a login.
'connect_url': pipeline.get_login_url(
state.provider.provider_id,
pipeline.AUTH_ENTRY_ACCOUNT_SETTINGS,
# The url the user should be directed to after the auth process has completed.
redirect_url=reverse('account_settings'),
),
'accepts_logins': state.provider.accepts_logins,
# If the user is connected, sending a POST request to this url removes the connection
# information for this provider from their edX account.
'disconnect_url': pipeline.get_disconnect_url(state.provider.provider_id, state.association_id),
# We only want to include providers if they are either currently available to be logged
# in with, or if the user is already authenticated with them.
} for state in auth_states if state.provider.display_for_login or state.has_account]
return context