Andal.LND/edx-platform
4
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #19693 from edx/mdikan/hackathon-21

Browse Source 
Removal of deprecated external auth.
This commit is contained in:
Mike Dikan
2019-03-04 13:48:57 -05:00
committed by GitHub
parent c8ebcff16b 2c149ca6be
commit 2fc27a9caa
45 changed files with 119 additions and 3660 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
openedx/core/djangoapps/external_auth/__init__.py
View File

17
openedx/core/djangoapps/external_auth/admin.py
View File

@@ -1,17 +0,0 @@
'''
django admin pages for courseware model
'''
from django.contrib import admin
from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
class ExternalAuthMapAdmin(admin.ModelAdmin):
"""
Admin model for ExternalAuthMap
"""
search_fields = ['external_id', 'user__username']
date_hierarchy = 'dtcreated'
admin.site.register(ExternalAuthMap, ExternalAuthMapAdmin)

131
openedx/core/djangoapps/external_auth/djangostore.py
View File

@@ -1,131 +0,0 @@
"""A openid store using django cache"""
import logging
import time
from django.core.cache import cache
from openid.store import nonce
from openid.store.interface import OpenIDStore
DEFAULT_ASSOCIATIONS_TIMEOUT = 60
DEFAULT_NONCE_TIMEOUT = 600
ASSOCIATIONS_KEY_PREFIX = 'openid.provider.associations.'
NONCE_KEY_PREFIX = 'openid.provider.nonce.'
log = logging.getLogger('DjangoOpenIDStore')
def get_url_key(server_url):
"""
Returns the URL key for the given server_url.
"""
return ASSOCIATIONS_KEY_PREFIX + server_url
def get_nonce_key(server_url, timestamp, salt):
"""
Returns the nonce for the given parameters.
"""
return '{prefix}{url}.{ts}.{salt}'.format(
prefix=NONCE_KEY_PREFIX,
url=server_url,
ts=timestamp,
salt=salt,
)
class DjangoOpenIDStore(OpenIDStore):
"""
django implementation of OpenIDStore.
"""
def __init__(self):
log.info('DjangoStore cache:' + str(cache.__class__))
def storeAssociation(self, server_url, assoc):
key = get_url_key(server_url)
log.info(u'storeAssociation {0}'.format(key))
associations = cache.get(key, {})
associations[assoc.handle] = assoc
cache.set(key, associations, DEFAULT_ASSOCIATIONS_TIMEOUT)
def getAssociation(self, server_url, handle=None):
key = get_url_key(server_url)
log.info(u'getAssociation {0}'.format(key))
associations = cache.get(key, {})
assoc = None
if handle is None:
# get best association
valid_assocs = [a for a in associations if a.getExpiresIn() > 0]
if valid_assocs:
valid_assocs.sort(lambda a: a.getExpiresIn(), reverse=True)
assoc = valid_assocs.sort[0]
else:
assoc = associations.get(handle)
# check expiration and remove if it has expired
if assoc and assoc.getExpiresIn() <= 0:
if handle is None:
cache.delete(key)
else:
associations.pop(handle)
cache.set(key, associations, DEFAULT_ASSOCIATIONS_TIMEOUT)
assoc = None
return assoc
def removeAssociation(self, server_url, handle):
key = get_url_key(server_url)
log.info(u'removeAssociation {0}'.format(key))
associations = cache.get(key, {})
removed = False
if associations:
if handle is None:
cache.delete(key)
removed = True
else:
assoc = associations.pop(handle, None)
if assoc:
cache.set(key, associations, DEFAULT_ASSOCIATIONS_TIMEOUT)
removed = True
return removed
def useNonce(self, server_url, timestamp, salt):
key = get_nonce_key(server_url, timestamp, salt)
log.info(u'useNonce {0}'.format(key))
if abs(timestamp - time.time()) > nonce.SKEW:
return False
anonce = cache.get(key)
found = False
if anonce is None:
cache.set(key, '-', DEFAULT_NONCE_TIMEOUT)
found = False
else:
found = True
return found
def cleanupNonces(self):
# not necesary, keys will timeout
return 0
def cleanupAssociations(self):
# not necesary, keys will timeout
return 0

99
openedx/core/djangoapps/external_auth/login_and_register.py
View File

@@ -1,99 +0,0 @@
"""Intercept login and registration requests.
This module contains legacy code originally from `student.views`.
"""
import re
from django.conf import settings
from django.urls import reverse
from django.shortcuts import redirect
from opaque_keys.edx.keys import CourseKey
from six import text_type
import openedx.core.djangoapps.external_auth.views
from xmodule.modulestore.django import modulestore
# pylint: disable=fixme
# TODO: This function is kind of gnarly/hackish/etc and is only used in one location.
# It'd be awesome if we could get rid of it; manually parsing course_id strings form larger strings
# seems Probably Incorrect
def _parse_course_id_from_string(input_str):
"""
Helper function to determine if input_str (typically the queryparam 'next') contains a course_id.
@param input_str:
@return: the course_id if found, None if not
"""
m_obj = re.match(r'^/courses/{}'.format(settings.COURSE_ID_PATTERN), input_str)
if m_obj:
return CourseKey.from_string(m_obj.group('course_id'))
return None
def _get_course_enrollment_domain(course_id):
"""
Helper function to get the enrollment domain set for a course with id course_id
@param course_id:
@return:
"""
course = modulestore().get_course(course_id)
if course is None:
return None
return course.enrollment_domain
def login(request):
"""Allow external auth to intercept and handle a login request.
Arguments:
request (Request): A request for the login page.
Returns:
Response or None
"""
# Default to a `None` response, indicating that external auth
# is not handling the request.
response = None
if (
settings.FEATURES['AUTH_USE_CERTIFICATES'] and
openedx.core.djangoapps.external_auth.views.ssl_get_cert_from_request(request)
):
# SSL login doesn't require a view, so redirect
# branding and allow that to process the login if it
# is enabled and the header is in the request.
response = openedx.core.djangoapps.external_auth.views.redirect_with_get('root', request.GET)
elif settings.FEATURES.get('AUTH_USE_CAS'):
# If CAS is enabled, redirect auth handling to there
response = redirect(reverse('cas-login'))
elif settings.FEATURES.get('AUTH_USE_SHIB'):
redirect_to = request.GET.get('next')
if redirect_to:
course_id = _parse_course_id_from_string(redirect_to)
if course_id and _get_course_enrollment_domain(course_id):
response = openedx.core.djangoapps.external_auth.views.course_specific_login(
request,
text_type(course_id),
)
return response
def register(request):
"""Allow external auth to intercept and handle a registration request.
Arguments:
request (Request): A request for the registration page.
Returns:
Response or None
"""
response = None
if settings.FEATURES.get('AUTH_USE_CERTIFICATES_IMMEDIATE_SIGNUP'):
# Redirect to branding to process their certificate if SSL is enabled
# and registration is disabled.
response = openedx.core.djangoapps.external_auth.views.redirect_with_get('root', request.GET)
return response

34
openedx/core/djangoapps/external_auth/migrations/0001_initial.py
View File

@@ -1,34 +0,0 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
from django.conf import settings
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='ExternalAuthMap',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('external_id', models.CharField(max_length=255, db_index=True)),
('external_domain', models.CharField(max_length=255, db_index=True)),
('external_credentials', models.TextField(blank=True)),
('external_email', models.CharField(max_length=255, db_index=True)),
('external_name', models.CharField(db_index=True, max_length=255, blank=True)),
('internal_password', models.CharField(max_length=31, blank=True)),
('dtcreated', models.DateTimeField(auto_now_add=True, verbose_name=b'creation date')),
('dtsignup', models.DateTimeField(null=True, verbose_name=b'signup date')),
('user', models.OneToOneField(null=True, to=settings.AUTH_USER_MODEL, on_delete=models.CASCADE)),
],
),
migrations.AlterUniqueTogether(
name='externalauthmap',
unique_together=set([('external_id', 'external_domain')]),
),
]

0
openedx/core/djangoapps/external_auth/migrations/__init__.py
View File

39
openedx/core/djangoapps/external_auth/models.py
View File

@@ -1,39 +0,0 @@
"""
WE'RE USING MIGRATIONS!
If you make changes to this model, be sure to create an appropriate migration
file and check it in at the same time as your model changes. To do that,
1. Go to the edx-platform dir
2. ./manage.py lms schemamigration student --auto description_of_your_change
3. Add the migration file created in edx-platform/openedx/core/djangoapps/external_auth/migrations/
"""
from django.contrib.auth.models import User
from django.db import models
class ExternalAuthMap(models.Model):
"""
Model class for external auth.
.. pii: Contains PII used in mapping external auth. Unused and empty on edx.org.
.. pii_types: name, email_address, password, external_service
.. pii_retirement: retained
"""
class Meta(object):
app_label = "external_auth"
unique_together = (('external_id', 'external_domain'), )
external_id = models.CharField(max_length=255, db_index=True)
external_domain = models.CharField(max_length=255, db_index=True)
external_credentials = models.TextField(blank=True) # JSON dictionary
external_email = models.CharField(max_length=255, db_index=True)
external_name = models.CharField(blank=True, max_length=255, db_index=True)
user = models.OneToOneField(User, unique=True, db_index=True, null=True, on_delete=models.CASCADE)
internal_password = models.CharField(blank=True, max_length=31) # randomly generated
dtcreated = models.DateTimeField('creation date', auto_now_add=True)
dtsignup = models.DateTimeField('signup date', null=True) # set after signup
def __unicode__(self):
return "[%s] = (%s / %s)" % (self.external_id, self.external_name, self.external_email)

0
openedx/core/djangoapps/external_auth/tests/__init__.py
View File

28
openedx/core/djangoapps/external_auth/tests/test_helper.py
View File

@@ -1,28 +0,0 @@
"""
Tests for utility functions in external_auth module
"""
from django.test import TestCase
from openedx.core.djangoapps.external_auth.views import _safe_postlogin_redirect
class ExternalAuthHelperFnTest(TestCase):
"""
Unit tests for the external_auth.views helper function
"""
def test__safe_postlogin_redirect(self):
"""
Tests the _safe_postlogin_redirect function with different values of next
"""
HOST = 'testserver' # pylint: disable=invalid-name
ONSITE1 = '/dashboard' # pylint: disable=invalid-name
ONSITE2 = '/courses/org/num/name/courseware' # pylint: disable=invalid-name
ONSITE3 = 'http://{}/my/custom/url'.format(HOST) # pylint: disable=invalid-name
OFFSITE1 = 'http://www.attacker.com' # pylint: disable=invalid-name
for redirect_to in [ONSITE1, ONSITE2, ONSITE3, OFFSITE1]:
redir = _safe_postlogin_redirect(redirect_to, HOST)
self.assertEqual(redir.status_code, 302)
if redirect_to in [ONSITE3, OFFSITE1]:
self.assertEqual(redir['location'], "/")
else:
self.assertEqual(redir['location'], redirect_to)

474
openedx/core/djangoapps/external_auth/tests/test_openid_provider.py
View File

@@ -1,474 +0,0 @@
#-*- encoding=utf-8 -*-
'''
Created on Jan 18, 2013
@author: brian
'''
from __future__ import print_function
import openid
from openid.fetchers import HTTPFetcher, HTTPResponse
from urlparse import parse_qs, urlparse
from django.conf import settings
from django.test import TestCase, LiveServerTestCase
from django.core.cache import cache
from django.test.utils import override_settings
from django.urls import reverse
from django.test.client import RequestFactory
from unittest import skipUnless
from student.tests.factories import UserFactory
from openedx.core.djangoapps.external_auth.views import provider_login
class MyFetcher(HTTPFetcher):
"""A fetcher that uses server-internal calls for performing HTTP
requests.
"""
def __init__(self, client):
"""@param client: A test client object"""
super(MyFetcher, self).__init__()
self.client = client
def fetch(self, url, body=None, headers=None):
"""Perform an HTTP request
@raises Exception: Any exception that can be raised by Django
@see: C{L{HTTPFetcher.fetch}}
"""
if body:
# method = 'POST'
# undo the URL encoding of the POST arguments
data = parse_qs(body)
response = self.client.post(url, data)
else:
# method = 'GET'
data = {}
if headers and 'Accept' in headers:
data['CONTENT_TYPE'] = headers['Accept']
response = self.client.get(url, data)
# Translate the test client response to the fetcher's HTTP response abstraction
content = response.content
final_url = url
response_headers = {}
if 'Content-Type' in response:
response_headers['content-type'] = response['Content-Type']
if 'X-XRDS-Location' in response:
response_headers['x-xrds-location'] = response['X-XRDS-Location']
status = response.status_code
return HTTPResponse(
body=content,
final_url=final_url,
headers=response_headers,
status=status,
)
class OpenIdProviderTest(TestCase):
"""
Tests of the OpenId login
"""
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_begin_login_with_xrds_url(self):
# the provider URL must be converted to an absolute URL in order to be
# used as an openid provider.
provider_url = reverse('openid-provider-xrds')
factory = RequestFactory()
request = factory.request()
abs_provider_url = request.build_absolute_uri(location=provider_url)
# In order for this absolute URL to work (i.e. to get xrds, then authentication)
# in the test environment, we either need a live server that works with the default
# fetcher (i.e. urlopen2), or a test server that is reached through a custom fetcher.
# Here we do the latter:
fetcher = MyFetcher(self.client)
openid.fetchers.setDefaultFetcher(fetcher, wrap_exceptions=False)
# now we can begin the login process by invoking a local openid client,
# with a pointer to the (also-local) openid provider:
with self.settings(OPENID_SSO_SERVER_URL=abs_provider_url):
url = reverse('openid-login')
resp = self.client.post(url)
code = 200
self.assertEqual(resp.status_code, code,
u"got code {0} for url '{1}'. Expected code {2}"
.format(resp.status_code, url, code))
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_begin_login_with_login_url(self):
# the provider URL must be converted to an absolute URL in order to be
# used as an openid provider.
provider_url = reverse('openid-provider-login')
factory = RequestFactory()
request = factory.request()
abs_provider_url = request.build_absolute_uri(location=provider_url)
# In order for this absolute URL to work (i.e. to get xrds, then authentication)
# in the test environment, we either need a live server that works with the default
# fetcher (i.e. urlopen2), or a test server that is reached through a custom fetcher.
# Here we do the latter:
fetcher = MyFetcher(self.client)
openid.fetchers.setDefaultFetcher(fetcher, wrap_exceptions=False)
# now we can begin the login process by invoking a local openid client,
# with a pointer to the (also-local) openid provider:
with self.settings(OPENID_SSO_SERVER_URL=abs_provider_url):
url = reverse('openid-login')
resp = self.client.post(url)
code = 200
self.assertEqual(resp.status_code, code,
u"got code {0} for url '{1}'. Expected code {2}"
.format(resp.status_code, url, code))
for expected_input in (
'<input name="openid.ns" type="hidden" value="http://specs.openid.net/auth/2.0" />',
'<input name="openid.ns.ax" type="hidden" value="http://openid.net/srv/ax/1.0" />',
'<input name="openid.ax.type.fullname" type="hidden" value="http://axschema.org/namePerson" />',
'<input type="submit" value="Continue" />',
'<input name="openid.ax.type.email" type="hidden" value="http://axschema.org/contact/email" />',
'<input name="openid.ax.type.lastname" '
'type="hidden" value="http://axschema.org/namePerson/last" />',
'<input name="openid.ax.type.firstname" '
'type="hidden" value="http://axschema.org/namePerson/first" />',
'<input name="openid.ax.required" type="hidden" '
'value="email,fullname,old_email,firstname,old_nickname,lastname,old_fullname,nickname" />',
'<input name="openid.ax.type.nickname" '
'type="hidden" value="http://axschema.org/namePerson/friendly" />',
'<input name="openid.ax.type.old_email" '
'type="hidden" value="http://schema.openid.net/contact/email" />',
'<input name="openid.ax.type.old_nickname" '
'type="hidden" value="http://schema.openid.net/namePerson/friendly" />',
'<input name="openid.ax.type.old_fullname" '
'type="hidden" value="http://schema.openid.net/namePerson" />',
'<input name="openid.identity" '
'type="hidden" value="http://specs.openid.net/auth/2.0/identifier_select" />',
'<input name="openid.claimed_id" '
'type="hidden" value="http://specs.openid.net/auth/2.0/identifier_select" />',
# should work on the test server as well
'<input name="openid.realm" '
'type="hidden" value="http://testserver/" />',
):
self.assertContains(resp, expected_input, html=True)
# not included here are elements that will vary from run to run:
# <input name="openid.return_to" type="hidden"
# value="http://testserver/openid/complete/?janrain_nonce=2013-01-23T06%3A20%3A17ZaN7j6H" />
# <input name="openid.assoc_handle" type="hidden" value="{HMAC-SHA1}{50ff8120}{rh87+Q==}" />
def attempt_login(self, expected_code, login_method='POST', **kwargs):
""" Attempt to log in through the open id provider login """
url = reverse('openid-provider-login')
args = {
"openid.mode": "checkid_setup",
"openid.return_to": "http://testserver/openid/complete/?janrain_nonce=2013-01-23T06%3A20%3A17ZaN7j6H",
"openid.assoc_handle": "{HMAC-SHA1}{50ff8120}{rh87+Q==}",
"openid.claimed_id": "http://specs.openid.net/auth/2.0/identifier_select",
"openid.ns": "http://specs.openid.net/auth/2.0",
"openid.realm": "http://testserver/",
"openid.identity": "http://specs.openid.net/auth/2.0/identifier_select",
"openid.ns.ax": "http://openid.net/srv/ax/1.0",
"openid.ax.mode": "fetch_request",
"openid.ax.required": "email,fullname,old_email,firstname,old_nickname,lastname,old_fullname,nickname",
"openid.ax.type.fullname": "http://axschema.org/namePerson",
"openid.ax.type.lastname": "http://axschema.org/namePerson/last",
"openid.ax.type.firstname": "http://axschema.org/namePerson/first",
"openid.ax.type.nickname": "http://axschema.org/namePerson/friendly",
"openid.ax.type.email": "http://axschema.org/contact/email",
"openid.ax.type.old_email": "http://schema.openid.net/contact/email",
"openid.ax.type.old_nickname": "http://schema.openid.net/namePerson/friendly",
"openid.ax.type.old_fullname": "http://schema.openid.net/namePerson",
}
# override the default args with any given arguments
for key in kwargs:
args["openid." + key] = kwargs[key]
if login_method == 'POST':
resp = self.client.post(url, args)
elif login_method == 'GET':
resp = self.client.get(url, args)
else:
self.fail('Invalid login method')
code = expected_code
self.assertEqual(resp.status_code, code,
u"got code {0} for url '{1}'. Expected code {2}"
.format(resp.status_code, url, code))
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_open_id_setup(self):
""" Attempt a standard successful login """
self.attempt_login(200)
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_invalid_namespace(self):
""" Test for 403 error code when the namespace of the request is invalid"""
self.attempt_login(403, ns="http%3A%2F%2Fspecs.openid.net%2Fauth%2F2.0")
@override_settings(OPENID_PROVIDER_TRUSTED_ROOTS=['http://apps.cs50.edx.org'])
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_invalid_return_url(self):
""" Test for 403 error code when the url"""
self.attempt_login(403, return_to="http://apps.cs50.edx.or")
def _send_bad_redirection_login(self):
"""
Attempt to log in to the provider with setup parameters
Intentionally fail the login to force a redirect
"""
user = UserFactory()
factory = RequestFactory()
post_params = {'email': user.email, 'password': 'password'}
fake_url = 'fake url'
request = factory.post(reverse('openid-provider-login'), post_params)
openid_setup = {
'request': factory.request(),
'url': fake_url,
'post_params': {}
}
request.session = {
'openid_setup': openid_setup
}
response = provider_login(request)
return response
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_login_openid_handle_redirection(self):
""" Test to see that we can handle login redirection properly"""
response = self._send_bad_redirection_login()
self.assertEquals(response.status_code, 302)
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_login_openid_handle_redirection_ratelimited(self):
# try logging in 30 times, the default limit in the number of failed
# log in attempts before the rate gets limited
for _ in xrange(30):
self._send_bad_redirection_login()
response = self._send_bad_redirection_login()
# verify that we are not returning the default 403
self.assertEquals(response.status_code, 302)
# clear the ratelimit cache so that we don't fail other logins
cache.clear()
def _attempt_login_and_perform_final_response(self, user, profile_name):
"""
Performs full procedure of a successful OpenID provider login for user,
all required data is taken form ``user`` attribute which is an instance
of ``User`` model. As a convenience this method will also set
``profile.name`` for the user.
"""
url = reverse('openid-provider-login')
# login to the client so that we can persist session information
user.profile.name = profile_name
user.profile.save()
# It is asssumed that user's password is test (default for UserFactory)
self.client.login(username=user.username, password='test')
# login once to get the right session information
self.attempt_login(200)
post_args = {
'email': user.email,
'password': 'test'
}
# call url again, this time with username and password
return self.client.post(url, post_args)
@skipUnless(
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'), 'OpenID not enabled')
def test_provider_login_can_handle_unicode_email(self):
user = UserFactory(email=u"user.ąęł@gmail.com")
resp = self._attempt_login_and_perform_final_response(user, u"Jan ĄĘŁ")
location = resp['Location']
parsed_url = urlparse(location)
parsed_qs = parse_qs(parsed_url.query)
self.assertEquals(parsed_qs['openid.ax.type.ext1'][0], 'http://axschema.org/contact/email')
self.assertEquals(parsed_qs['openid.ax.type.ext0'][0], 'http://axschema.org/namePerson')
self.assertEquals(parsed_qs['openid.ax.value.ext0.1'][0],
user.profile.name.encode('utf-8'))
self.assertEquals(parsed_qs['openid.ax.value.ext1.1'][0],
user.email.encode('utf-8')) # pylint: disable=no-member
@skipUnless(
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'), 'OpenID not enabled')
def test_provider_login_can_handle_unicode_email_invalid_password(self):
user = UserFactory(email=u"user.ąęł@gmail.com")
url = reverse('openid-provider-login')
# login to the client so that we can persist session information
user.profile.name = u"Jan ĄĘ"
user.profile.save()
# It is asssumed that user's password is test (default for UserFactory)
self.client.login(username=user.username, password='test')
# login once to get the right session information
self.attempt_login(200)
# We trigger situation where user password is invalid at last phase
# of openid login
post_args = {
'email': user.email,
'password': 'invalid-password'
}
# call url again, this time with username and password
return self.client.post(url, post_args)
@skipUnless(
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'), 'OpenID not enabled')
def test_provider_login_can_handle_unicode_email_inactive_account(self):
user = UserFactory(email=u"user.ąęł@gmail.com", username=u"ąęół")
url = reverse('openid-provider-login')
# login to the client so that we can persist session information
user.profile.name = u'Jan ĄĘ'
user.profile.save() # pylint: disable=no-member
self.client.login(username=user.username, password='test')
# login once to get the right session information
self.attempt_login(200)
# We trigger situation where user is not active at final phase of
# OpenId login.
user.is_active = False
user.save()
post_args = {
'email': user.email,
'password': 'test'
}
# call url again, this time with username and password
self.client.post(url, post_args)
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_openid_final_response(self):
user = UserFactory()
# login to the client so that we can persist session information
for name in ['Robot 33', '☃']:
resp = self._attempt_login_and_perform_final_response(user, name)
# all information is embedded in the redirect url
location = resp['Location']
# parse the url
parsed_url = urlparse(location)
parsed_qs = parse_qs(parsed_url.query)
self.assertEquals(parsed_qs['openid.ax.type.ext1'][0], 'http://axschema.org/contact/email')
self.assertEquals(parsed_qs['openid.ax.type.ext0'][0], 'http://axschema.org/namePerson')
self.assertEquals(parsed_qs['openid.ax.value.ext1.1'][0], user.email)
self.assertEquals(parsed_qs['openid.ax.value.ext0.1'][0], user.profile.name)
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_openid_invalid_password(self):
url = reverse('openid-provider-login')
user = UserFactory()
# login to the client so that we can persist session information
for method in ['POST', 'GET']:
self.client.login(username=user.username, password='test')
self.attempt_login(200, method)
openid_setup = self.client.session['openid_setup']
self.assertIn('post_params', openid_setup)
post_args = {
'email': user.email,
'password': 'bad_password',
}
# call url again, this time with username and password
resp = self.client.post(url, post_args)
self.assertEquals(resp.status_code, 302)
redirect_url = resp['Location']
parsed_url = urlparse(redirect_url)
query_params = parse_qs(parsed_url[4])
self.assertIn('openid.return_to', query_params)
self.assertTrue(
query_params['openid.return_to'][0].startswith('http://testserver/openid/complete/')
)
class OpenIdProviderLiveServerTest(LiveServerTestCase):
"""
In order for this absolute URL to work (i.e. to get xrds, then authentication)
in the test environment, we either need a live server that works with the default
fetcher (i.e. urlopen2), or a test server that is reached through a custom fetcher.
Here we do the former.
"""
@skipUnless(settings.FEATURES.get('AUTH_USE_OPENID') and
settings.FEATURES.get('AUTH_USE_OPENID_PROVIDER'),
'OpenID not enabled')
def test_begin_login(self):
# the provider URL must be converted to an absolute URL in order to be
# used as an openid provider.
provider_url = reverse('openid-provider-xrds')
factory = RequestFactory()
request = factory.request()
abs_provider_url = request.build_absolute_uri(location=provider_url)
# In order for this absolute URL to work (i.e. to get xrds, then authentication)
# in the test environment, we either need a live server that works with the default
# fetcher (i.e. urlopen2), or a test server that is reached through a custom fetcher.
# Here we do the latter:
fetcher = MyFetcher(self.client)
openid.fetchers.setDefaultFetcher(fetcher, wrap_exceptions=False)
# now we can begin the login process by invoking a local openid client,
# with a pointer to the (also-local) openid provider:
with self.settings(OPENID_SSO_SERVER_URL=abs_provider_url):
url = reverse('openid-login')
resp = self.client.post(url)
code = 200
self.assertEqual(resp.status_code, code,
u"got code {0} for url '{1}'. Expected code {2}"
.format(resp.status_code, url, code))
@classmethod
def tearDownClass(cls):
"""
Workaround for a runtime error that occurs
intermittently when the server thread doesn't shut down
within 2 seconds.
Since the server is running in a Django thread and will
be terminated when the test suite terminates,
this shouldn't cause a resource allocation issue.
"""
try:
super(OpenIdProviderLiveServerTest, cls).tearDownClass()
except RuntimeError:
print("Warning: Could not shut down test server.")

601
openedx/core/djangoapps/external_auth/tests/test_shib.py
View File

@@ -1,601 +0,0 @@
# -*- coding: utf-8 -*-
"""
Tests for Shibboleth Authentication
@jbau
"""
import unittest
from importlib import import_module
from urllib import urlencode
from ddt import ddt, data
from django.conf import settings
from django.http import HttpResponseRedirect
from django.test import TestCase
from django.test.client import RequestFactory, Client as DjangoTestClient
from django.test.utils import override_settings
from django.urls import reverse
from django.contrib.auth.models import AnonymousUser, User
from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
from openedx.core.djangoapps.external_auth.views import (
shib_login, course_specific_login, course_specific_register, _flatten_to_ascii
)
from openedx.core.djangoapps.user_api import accounts as accounts_settings
from mock import patch
from six import text_type
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
from student.views import change_enrollment
from student.models import UserProfile, CourseEnrollment
from student.tests.factories import UserFactory
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore import ModuleStoreEnum
# Shib is supposed to provide 'REMOTE_USER', 'givenName', 'sn', 'mail', 'Shib-Identity-Provider'
# attributes via request.META. We can count on 'Shib-Identity-Provider', and 'REMOTE_USER' being present
# b/c of how mod_shib works but should test the behavior with the rest of the attributes present/missing
# For the sake of python convention we'll make all of these variable names ALL_CAPS
# These values would all returned from request.META, so they need to be str, not unicode
IDP = 'https://idp.stanford.edu/'
REMOTE_USER = 'test_user@stanford.edu'
MAILS = [None, '', 'test_user@stanford.edu'] # unicode shouldn't be in emails, would fail django's email validator
DISPLAYNAMES = [None, '', 'Jason 包']
GIVENNAMES = [None, '', 'jasön; John; bob'] # At Stanford, the givenNames can be a list delimited by ';'
SNS = [None, '', '包; smith'] # At Stanford, the sns can be a list delimited by ';'
def gen_all_identities():
"""
A generator for all combinations of test inputs.
Each generated item is a dict that represents what a shib IDP
could potentially pass to django via request.META, i.e.
setting (or not) request.META['givenName'], etc.
"""
def _build_identity_dict(mail, display_name, given_name, surname):
""" Helper function to return a dict of test identity """
meta_dict = {'Shib-Identity-Provider': IDP,
'REMOTE_USER': REMOTE_USER}
if display_name is not None:
meta_dict['displayName'] = display_name
if mail is not None:
meta_dict['mail'] = mail
if given_name is not None:
meta_dict['givenName'] = given_name
if surname is not None:
meta_dict['sn'] = surname
return meta_dict
for mail in MAILS:
for given_name in GIVENNAMES:
for surname in SNS:
for display_name in DISPLAYNAMES:
yield _build_identity_dict(mail, display_name, given_name, surname)
@ddt
@override_settings(SESSION_ENGINE='django.contrib.sessions.backends.cache')
class ShibSPTest(CacheIsolationTestCase):
"""
Tests for the Shibboleth SP, which communicates via request.META
(Apache environment variables set by mod_shib)
"""
ENABLED_CACHES = ['default']
request_factory = RequestFactory()
def setUp(self):
super(ShibSPTest, self).setUp()
self.test_user_id = ModuleStoreEnum.UserID.test
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
def test_exception_shib_login(self):
"""
Tests that we get the error page when there is no REMOTE_USER
or Shib-Identity-Provider in request.META
"""
no_remote_user_response = self.client.get(reverse('shib-login'), HTTP_SHIB_IDENTITY_PROVIDER=IDP)
self.assertEqual(no_remote_user_response.status_code, 403)
self.assertIn("identity server did not return your ID information", no_remote_user_response.content)
no_idp_response = self.client.get(reverse('shib-login'), HTTP_REMOTE_USER=REMOTE_USER)
self.assertEqual(no_idp_response.status_code, 403)
self.assertIn("identity server did not return your ID information", no_idp_response.content)
def _assert_shib_login_is_logged(self, audit_log_call, remote_user):
"""Asserts that shibboleth login attempt is being logged"""
remote_user = _flatten_to_ascii(remote_user) # django usernames have to be ascii
method_name, args, _kwargs = audit_log_call
self.assertEquals(method_name, 'info')
self.assertEquals(len(args), 1)
self.assertIn(u'logged in via Shibboleth', args[0])
self.assertIn(remote_user, args[0])
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
def test_shib_login(self):
"""
Tests that:
* shib credentials that match an existing ExternalAuthMap with a linked active user logs the user in
* shib credentials that match an existing ExternalAuthMap with a linked inactive user shows error page
* shib credentials that match an existing ExternalAuthMap without a linked user and also match the email
of an existing user without an existing ExternalAuthMap links the two and log the user in
* shib credentials that match an existing ExternalAuthMap without a linked user and also match the email
of an existing user that already has an ExternalAuthMap causes an error (403)
* shib credentials that do not match an existing ExternalAuthMap causes the registration form to appear
"""
# pylint: disable=too-many-statements
user_w_map = UserFactory.create(email='withmap@stanford.edu')
extauth = ExternalAuthMap(external_id='withmap@stanford.edu',
external_email='',
external_domain='shib:https://idp.stanford.edu/',
external_credentials="",
user=user_w_map)
user_wo_map = UserFactory.create(email='womap@stanford.edu')
user_w_map.save()
user_wo_map.save()
extauth.save()
inactive_user = UserFactory.create(email='inactive@stanford.edu')
inactive_user.is_active = False
inactive_extauth = ExternalAuthMap(external_id='inactive@stanford.edu',
external_email='',
external_domain='shib:https://idp.stanford.edu/',
external_credentials="",
user=inactive_user)
inactive_user.save()
inactive_extauth.save()
idps = ['https://idp.stanford.edu/', 'https://someother.idp.com/']
remote_users = ['withmap@stanford.edu', 'womap@stanford.edu',
'testuser2@someother_idp.com', 'inactive@stanford.edu']
for idp in idps:
for remote_user in remote_users:
self.client.logout()
with patch('openedx.core.djangoapps.external_auth.views.AUDIT_LOG') as mock_audit_log:
response = self.client.get(
reverse('shib-login'),
**{
'Shib-Identity-Provider': idp,
'mail': remote_user,
'REMOTE_USER': remote_user,
}
)
audit_log_calls = mock_audit_log.method_calls
if idp == "https://idp.stanford.edu/" and remote_user == 'withmap@stanford.edu':
self.assertRedirects(response, '/dashboard')
self.assertEquals(int(self.client.session['_auth_user_id']), user_w_map.id)
# verify logging:
self.assertEquals(len(audit_log_calls), 2)
self._assert_shib_login_is_logged(audit_log_calls[0], remote_user)
method_name, args, _kwargs = audit_log_calls[1]
self.assertEquals(method_name, 'info')
self.assertEquals(len(args), 1)
self.assertIn(u'Login success', args[0])
self.assertIn(remote_user, args[0])
elif idp == "https://idp.stanford.edu/" and remote_user == 'inactive@stanford.edu':
self.assertEqual(response.status_code, 403)
self.assertIn("Account not yet activated: please look for link in your email", response.content)
# verify logging:
self.assertEquals(len(audit_log_calls), 2)
self._assert_shib_login_is_logged(audit_log_calls[0], remote_user)
method_name, args, _kwargs = audit_log_calls[1]
self.assertEquals(method_name, 'warning')
self.assertEquals(len(args), 1)
self.assertIn(u'is not active after external login', args[0])
# self.assertEquals(remote_user, args[1])
elif idp == "https://idp.stanford.edu/" and remote_user == 'womap@stanford.edu':
self.assertIsNotNone(ExternalAuthMap.objects.get(user=user_wo_map))
self.assertRedirects(response, '/dashboard')
self.assertEquals(int(self.client.session['_auth_user_id']), user_wo_map.id)
# verify logging:
self.assertEquals(len(audit_log_calls), 2)
self._assert_shib_login_is_logged(audit_log_calls[0], remote_user)
method_name, args, _kwargs = audit_log_calls[1]
self.assertEquals(method_name, 'info')
self.assertEquals(len(args), 1)
self.assertIn(u'Login success', args[0])
self.assertIn(remote_user, args[0])
elif idp == "https://someother.idp.com/" and remote_user in \
['withmap@stanford.edu', 'womap@stanford.edu', 'inactive@stanford.edu']:
self.assertEqual(response.status_code, 403)
self.assertIn("You have already created an account using an external login", response.content)
# no audit logging calls
self.assertEquals(len(audit_log_calls), 0)
else:
self.assertEqual(response.status_code, 200)
self.assertContains(response,
(u"Preferences for {platform_name}"
.format(platform_name=settings.PLATFORM_NAME)))
# no audit logging calls
self.assertEquals(len(audit_log_calls), 0)
def _test_auto_activate_user_with_flag(self, log_user_string="inactive@stanford.edu"):
"""
Tests that FEATURES['BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH'] means extauth automatically
linked users, activates them, and logs them in
"""
inactive_user = UserFactory.create(email='inactive@stanford.edu')
if not log_user_string:
log_user_string = u"user.id: {}".format(inactive_user.id)
inactive_user.is_active = False
inactive_user.save()
request = self.request_factory.get('/shib-login')
request.session = import_module(settings.SESSION_ENGINE).SessionStore() # empty session
request.META.update({
'Shib-Identity-Provider': 'https://idp.stanford.edu/',
'REMOTE_USER': 'inactive@stanford.edu',
'mail': 'inactive@stanford.edu'
})
request.user = AnonymousUser()
with patch('openedx.core.djangoapps.external_auth.views.AUDIT_LOG') as mock_audit_log:
response = shib_login(request)
audit_log_calls = mock_audit_log.method_calls
# reload user from db, since the view function works via db side-effects
inactive_user = User.objects.get(id=inactive_user.id)
self.assertIsNotNone(ExternalAuthMap.objects.get(user=inactive_user))
self.assertTrue(inactive_user.is_active)
self.assertIsInstance(response, HttpResponseRedirect)
self.assertEqual(request.user, inactive_user)
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)
method_name, args, _kwargs = audit_log_calls[2]
self.assertEquals(method_name, 'info')
self.assertEquals(len(args), 1)
self.assertIn(u'Login success', args[0])
self.assertIn(log_user_string, args[0])
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@patch.dict(settings.FEATURES, {'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': True, 'SQUELCH_PII_IN_LOGS': False})
def test_extauth_auto_activate_user_with_flag_no_squelch(self):
"""
Wrapper to run base_test_extauth_auto_activate_user_with_flag with {'SQUELCH_PII_IN_LOGS': False}
"""
self._test_auto_activate_user_with_flag(log_user_string="inactive@stanford.edu")
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@patch.dict(settings.FEATURES, {'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': True, 'SQUELCH_PII_IN_LOGS': True})
def test_extauth_auto_activate_user_with_flag_squelch(self):
"""
Wrapper to run base_test_extauth_auto_activate_user_with_flag with {'SQUELCH_PII_IN_LOGS': True}
"""
self._test_auto_activate_user_with_flag(log_user_string=None)
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@data(*gen_all_identities())
def test_registration_form(self, identity):
"""
Tests the registration form showing up with the proper parameters.
Uses django test client for its session support
"""
client = DjangoTestClient()
# identity k/v pairs will show up in request.META
response = client.get(path='/shib-login/', data={}, follow=False, **identity)
self.assertEquals(response.status_code, 200)
mail_input_html = '<input class="" id="email" type="email" name="email"'
if not identity.get('mail'):
self.assertContains(response, mail_input_html)
else:
self.assertNotContains(response, mail_input_html)
sn_empty = not identity.get('sn')
given_name_empty = not identity.get('givenName')
displayname_empty = not identity.get('displayName')
fullname_input_html = '<input id="name" type="text" name="name"'
if sn_empty and given_name_empty and displayname_empty:
self.assertContains(response, fullname_input_html)
else:
self.assertNotContains(response, fullname_input_html)
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@data(*gen_all_identities())
def test_registration_form_submit(self, identity):
"""
Tests user creation after the registration form that pops is submitted. If there is no shib
ExternalAuthMap in the session, then the created user should take the username and email from the
request.
Uses django test client for its session support
"""
# First we pop the registration form
self.client.get(path='/shib-login/', data={}, follow=False, **identity)
# Then we have the user answer the registration form
# These are unicode because request.POST returns unicode
postvars = {'email': u'post_email@stanford.edu',
'username': u'post_username', # django usernames can't be unicode
'password': u'post_pássword',
'name': u'post_náme',
'terms_of_service': u'true',
'honor_code': u'true'}
with patch('openedx.core.djangoapps.user_authn.views.register.AUDIT_LOG') as mock_audit_log:
self.client.post('/create_account', data=postvars)
mail = identity.get('mail')
# verify logging of login happening during account creation:
audit_log_calls = mock_audit_log.method_calls
self.assertEquals(len(audit_log_calls), 3)
method_name, args, _kwargs = audit_log_calls[0]
self.assertEquals(method_name, 'info')
self.assertEquals(len(args), 2)
self.assertIn(u'User registered with external_auth', args[0])
self.assertEquals(u'post_username', args[1])
method_name, args, _kwargs = audit_log_calls[1]
self.assertEquals(method_name, 'info')
self.assertEquals(len(args), 3)
self.assertIn(u'Updated ExternalAuthMap for ', args[0])
self.assertEquals(u'post_username', args[1])
self.assertEquals(u'test_user@stanford.edu', args[2].external_id)
method_name, args, _kwargs = audit_log_calls[2]
self.assertEquals(method_name, 'info')
self.assertEquals(len(args), 1)
self.assertIn(u'Login success on new account creation', args[0])
self.assertIn(u'post_username', args[0])
user = User.objects.get(id=self.client.session['_auth_user_id'])
# check that the created user has the right email, either taken from shib or user input
if mail:
self.assertEqual(user.email, mail)
self.assertEqual(list(User.objects.filter(email=postvars['email'])), [])
self.assertIsNotNone(User.objects.get(email=mail)) # get enforces only 1 such user
else:
self.assertEqual(user.email, postvars['email'])
self.assertEqual(list(User.objects.filter(email=mail)), [])
self.assertIsNotNone(User.objects.get(email=postvars['email'])) # get enforces only 1 such user
# check that the created user profile has the right name, either taken from shib or user input
profile = UserProfile.objects.get(user=user)
external_name = self.client.session['ExternalAuthMap'].external_name
displayname_empty = not identity.get('displayName')
if displayname_empty:
if len(external_name.strip()) < accounts_settings.NAME_MIN_LENGTH:
self.assertEqual(profile.name, postvars['name'])
else:
self.assertEqual(profile.name, external_name.strip())
self.assertNotIn(u';', profile.name)
else:
self.assertEqual(profile.name, self.client.session['ExternalAuthMap'].external_name)
self.assertEqual(profile.name, identity.get('displayName').decode('utf-8'))
@ddt
@override_settings(SESSION_ENGINE='django.contrib.sessions.backends.cache')
class ShibSPTestModifiedCourseware(ModuleStoreTestCase):
"""
Tests for the Shibboleth SP which modify the courseware
"""
ENABLED_CACHES = ['default', 'mongo_metadata_inheritance', 'loc_cache']
request_factory = RequestFactory()
def setUp(self):
super(ShibSPTestModifiedCourseware, self).setUp()
self.test_user_id = ModuleStoreEnum.UserID.test
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@data(None, "", "shib:https://idp.stanford.edu/")
def test_course_specific_login_and_reg(self, domain):
"""
Tests that the correct course specific login and registration urls work for shib
"""
course = CourseFactory.create(
org='MITx',
number='999',
display_name='Robot Super Course',
user_id=self.test_user_id,
)
# Test for cases where course is found
# set domains
# temporarily set the branch to draft-preferred so we can update the course
with self.store.branch_setting(ModuleStoreEnum.Branch.draft_preferred, course.id):
course.enrollment_domain = domain
self.store.update_item(course, self.test_user_id)
# setting location to test that GET params get passed through
login_request = self.request_factory.get('/course_specific_login/MITx/999/Robot_Super_Course' +
'?course_id=MITx/999/Robot_Super_Course' +
'&enrollment_action=enroll')
_reg_request = self.request_factory.get('/course_specific_register/MITx/999/Robot_Super_Course' +
'?course_id=MITx/999/course/Robot_Super_Course' +
'&enrollment_action=enroll')
login_response = course_specific_login(login_request, 'MITx/999/Robot_Super_Course')
reg_response = course_specific_register(login_request, 'MITx/999/Robot_Super_Course')
if domain and "shib" in domain:
self.assertIsInstance(login_response, HttpResponseRedirect)
self.assertEqual(login_response['Location'],
reverse('shib-login') +
'?course_id=MITx/999/Robot_Super_Course' +
'&enrollment_action=enroll')
self.assertIsInstance(login_response, HttpResponseRedirect)
self.assertEqual(reg_response['Location'],
reverse('shib-login') +
'?course_id=MITx/999/Robot_Super_Course' +
'&enrollment_action=enroll')
else:
self.assertIsInstance(login_response, HttpResponseRedirect)
self.assertEqual(login_response['Location'],
reverse('signin_user') +
'?course_id=MITx/999/Robot_Super_Course' +
'&enrollment_action=enroll')
self.assertIsInstance(login_response, HttpResponseRedirect)
self.assertEqual(reg_response['Location'],
reverse('register_user') +
'?course_id=MITx/999/Robot_Super_Course' +
'&enrollment_action=enroll')
# Now test for non-existent course
# setting location to test that GET params get passed through
login_request = self.request_factory.get('/course_specific_login/DNE/DNE/DNE' +
'?course_id=DNE/DNE/DNE' +
'&enrollment_action=enroll')
_reg_request = self.request_factory.get('/course_specific_register/DNE/DNE/DNE' +
'?course_id=DNE/DNE/DNE/Robot_Super_Course' +
'&enrollment_action=enroll')
login_response = course_specific_login(login_request, 'DNE/DNE/DNE')
reg_response = course_specific_register(login_request, 'DNE/DNE/DNE')
self.assertIsInstance(login_response, HttpResponseRedirect)
self.assertEqual(login_response['Location'],
reverse('signin_user') +
'?course_id=DNE/DNE/DNE' +
'&enrollment_action=enroll')
self.assertIsInstance(login_response, HttpResponseRedirect)
self.assertEqual(reg_response['Location'],
reverse('register_user') +
'?course_id=DNE/DNE/DNE' +
'&enrollment_action=enroll')
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
def test_enrollment_limit_by_domain(self):
"""
Tests that the enrollmentDomain setting is properly limiting enrollment to those who have
the proper external auth
"""
# create 2 course, one with limited enrollment one without
shib_course = CourseFactory.create(
org='Stanford',
number='123',
display_name='Shib Only',
enrollment_domain='shib:https://idp.stanford.edu/',
user_id=self.test_user_id,
)
open_enroll_course = CourseFactory.create(
org='MITx',
number='999',
display_name='Robot Super Course',
enrollment_domain='',
user_id=self.test_user_id,
)
# create 3 kinds of students, external_auth matching shib_course, external_auth not matching, no external auth
shib_student = UserFactory.create()
shib_student.save()
extauth = ExternalAuthMap(external_id='testuser@stanford.edu',
external_email='',
external_domain='shib:https://idp.stanford.edu/',
external_credentials="",
user=shib_student)
extauth.save()
other_ext_student = UserFactory.create()
other_ext_student.username = "teststudent2"
other_ext_student.email = "teststudent2@other.edu"
other_ext_student.save()
extauth = ExternalAuthMap(external_id='testuser1@other.edu',
external_email='',
external_domain='shib:https://other.edu/',
external_credentials="",
user=other_ext_student)
extauth.save()
int_student = UserFactory.create()
int_student.username = "teststudent3"
int_student.email = "teststudent3@gmail.com"
int_student.save()
# Tests the two case for courses, limited and not
for course in [shib_course, open_enroll_course]:
for student in [shib_student, other_ext_student, int_student]:
request = self.request_factory.post(
'/change_enrollment',
data={'enrollment_action': 'enroll', 'course_id': text_type(course.id)}
)
request.user = student
response = change_enrollment(request)
# If course is not limited or student has correct shib extauth then enrollment should be allowed
if course is open_enroll_course or student is shib_student:
self.assertEqual(response.status_code, 200)
self.assertTrue(CourseEnrollment.is_enrolled(student, course.id))
else:
self.assertEqual(response.status_code, 400)
self.assertFalse(CourseEnrollment.is_enrolled(student, course.id))
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
def test_shib_login_enrollment(self):
"""
A functionality test that a student with an existing shib login
can auto-enroll in a class with GET or POST params. Also tests the direction functionality of
the 'next' GET/POST param
"""
student = UserFactory.create()
extauth = ExternalAuthMap(external_id='testuser@stanford.edu',
external_email='',
external_domain='shib:https://idp.stanford.edu/',
external_credentials="",
internal_password="password",
user=student)
student.set_password("password")
student.save()
extauth.save()
course = CourseFactory.create(
org='Stanford',
number='123',
display_name='Shib Only',
enrollment_domain='shib:https://idp.stanford.edu/',
user_id=self.test_user_id,
)
# use django test client for sessions and url processing
# no enrollment before trying
self.assertFalse(CourseEnrollment.is_enrolled(student, course.id))
self.client.logout()
params = [
('course_id', text_type(course.id)),
('enrollment_action', 'enroll'),
('next', '/testredirect')
]
request_kwargs = {'path': '/shib-login/',
'data': dict(params),
'follow': False,
'REMOTE_USER': 'testuser@stanford.edu',
'Shib-Identity-Provider': 'https://idp.stanford.edu/',
'HTTP_ACCEPT': "text/html"}
response = self.client.get(**request_kwargs)
# successful login is a redirect to the URL that handles auto-enrollment
self.assertEqual(response.status_code, 302)
self.assertEqual(response['location'],
'/account/finish_auth?{}'.format(urlencode(params)))
class ShibUtilFnTest(TestCase):
"""
Tests util functions in shib module
"""
def test__flatten_to_ascii(self):
DIACRITIC = u"àèìòùÀÈÌÒÙáéíóúýÁÉÍÓÚÝâêîôûÂÊÎÔÛãñõÃÑÕäëïöüÿÄËÏÖÜŸåÅçÇ" # pylint: disable=invalid-name
STR_DIACRI = "àèìòùÀÈÌÒÙáéíóúýÁÉÍÓÚÝâêîôûÂÊÎÔÛãñõÃÑÕäëïöüÿÄËÏÖÜŸåÅçÇ" # pylint: disable=invalid-name
FLATTENED = u"aeiouAEIOUaeiouyAEIOUYaeiouAEIOUanoANOaeiouyAEIOUYaAcC" # pylint: disable=invalid-name
self.assertEqual(_flatten_to_ascii('jasön'), 'jason') # umlaut
self.assertEqual(_flatten_to_ascii('Jason包'), 'Jason') # mandarin, so it just gets dropped
self.assertEqual(_flatten_to_ascii('abc'), 'abc') # pass through
unicode_test = _flatten_to_ascii(DIACRITIC)
self.assertEqual(unicode_test, FLATTENED)
self.assertIsInstance(unicode_test, unicode)
str_test = _flatten_to_ascii(STR_DIACRI)
self.assertEqual(str_test, FLATTENED)
self.assertIsInstance(str_test, str)

424
openedx/core/djangoapps/external_auth/tests/test_ssl.py
View File

@@ -1,424 +0,0 @@
"""
Provides unit tests for SSL based authentication portions
of the external_auth app.
"""
from contextlib import contextmanager
import copy
from unittest import skip
from mock import Mock, patch
from six import text_type
from django.conf import settings
from django.contrib.auth import SESSION_KEY
from django.contrib.auth.models import AnonymousUser, User
from django.contrib.sessions.middleware import SessionMiddleware
from django.urls import reverse
from django.test.client import Client
from django.test.client import RequestFactory
from django.test.utils import override_settings
from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
import openedx.core.djangoapps.external_auth.views as external_auth_views
from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
from openedx.core.djangolib.testing.utils import skip_unless_cms, skip_unless_lms
from student.models import CourseEnrollment
from student.roles import CourseStaffRole
from student.tests.factories import UserFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
FEATURES_WITH_SSL_AUTH = settings.FEATURES.copy()
FEATURES_WITH_SSL_AUTH['AUTH_USE_CERTIFICATES'] = True
FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP = FEATURES_WITH_SSL_AUTH.copy()
FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP['AUTH_USE_CERTIFICATES_IMMEDIATE_SIGNUP'] = True
FEATURES_WITH_SSL_AUTH_AUTO_ACTIVATE = FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP.copy()
FEATURES_WITH_SSL_AUTH_AUTO_ACTIVATE['BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH'] = True
FEATURES_WITHOUT_SSL_AUTH = settings.FEATURES.copy()
FEATURES_WITHOUT_SSL_AUTH['AUTH_USE_CERTIFICATES'] = False
CACHES_ENABLE_GENERAL = copy.deepcopy(settings.CACHES)
CACHES_ENABLE_GENERAL['general']['BACKEND'] = 'django.core.cache.backends.locmem.LocMemCache'
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH)
@override_settings(CACHES=CACHES_ENABLE_GENERAL)
class SSLClientTest(ModuleStoreTestCase):
"""
Tests SSL Authentication code sections of external_auth
"""
AUTH_DN = b'/C=US/ST=Massachusetts/O=Massachusetts Institute of Technology/OU=Client CA v1/CN={0}/emailAddress={1}'
USER_NAME = 'test_user_ssl'
USER_EMAIL = 'test_user_ssl@EDX.ORG'
MOCK_URL = '/'
@contextmanager
def _create_ssl_request(self, url):
"""Creates a basic request for SSL use."""
request = self.factory.get(url)
request.META['SSL_CLIENT_S_DN'] = self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL)
request.site = SiteFactory.create()
request.user = AnonymousUser()
middleware = SessionMiddleware()
middleware.process_request(request)
request.session.save()
with patch('edxmako.request_context.get_current_request', return_value=request):
yield request
@contextmanager
def _create_normal_request(self, url):
"""Creates sessioned request without SSL headers"""
request = self.factory.get(url)
request.user = AnonymousUser()
middleware = SessionMiddleware()
middleware.process_request(request)
request.session.save()
with patch('edxmako.request_context.get_current_request', return_value=request):
yield request
def setUp(self):
"""Setup test case by adding primary user."""
super(SSLClientTest, self).setUp()
self.client = Client()
self.factory = RequestFactory()
self.mock = Mock()
@skip_unless_lms
def test_ssl_login_with_signup_lms(self):
"""
Validate that an SSL login creates an eamap user and
redirects them to the signup page.
"""
with self._create_ssl_request('/') as request:
response = external_auth_views.ssl_login(request)
# Response should contain template for signup form, eamap should have user, and internal
# auth should not have a user
self.assertIn('<form role="form" id="register-form" method="post"', response.content)
try:
ExternalAuthMap.objects.get(external_id=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to external auth map, exception was {0}'.format(str(ex)))
with self.assertRaises(User.DoesNotExist):
User.objects.get(email=self.USER_EMAIL)
@skip_unless_cms
def test_ssl_login_with_signup_cms(self):
"""
Validate that an SSL login creates an eamap user and
redirects them to the signup page on CMS.
"""
self.client.get(
reverse('login'),
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL)
)
try:
ExternalAuthMap.objects.get(external_id=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to external auth map, exception was {0}'.format(str(ex)))
with self.assertRaises(User.DoesNotExist):
User.objects.get(email=self.USER_EMAIL)
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_ssl_login_without_signup_lms(self):
"""
Test IMMEDIATE_SIGNUP feature flag and ensure the user account is automatically created
and the user is redirected to slash.
"""
with self._create_ssl_request('/') as request:
external_auth_views.ssl_login(request)
# Assert our user exists in both eamap and Users, and that we are logged in
try:
ExternalAuthMap.objects.get(external_id=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to external auth map, exception was {0}'.format(str(ex)))
try:
User.objects.get(email=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to internal users, exception was {0}'.format(str(ex)))
@skip_unless_cms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_ssl_login_without_signup_cms(self):
"""
Test IMMEDIATE_SIGNUP feature flag and ensure the user account is
automatically created on CMS, and that we are redirected
to courses.
"""
response = self.client.get(
reverse('login'),
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL)
)
self.assertEqual(response.status_code, 302)
self.assertIn('/course', response['location'])
# Assert our user exists in both eamap and Users, and that we are logged in
try:
ExternalAuthMap.objects.get(external_id=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to external auth map, exception was {0}'.format(str(ex)))
try:
User.objects.get(email=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to internal users, exception was {0}'.format(str(ex)))
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_default_login_decorator_ssl(self):
"""
Make sure that SSL login happens if it is enabled on protected
views instead of showing the login form.
"""
response = self.client.get(reverse('dashboard'), follows=True)
self.assertEqual(response.status_code, 302)
self.assertIn(reverse('signin_user'), response['location'])
response = self.client.get(
reverse('dashboard'), follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL))
self.assertEquals(('/dashboard', 302),
response.redirect_chain[-1])
self.assertIn(SESSION_KEY, self.client.session)
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_registration_page_bypass(self):
"""
This tests to make sure when immediate signup is on that
the user doesn't get presented with the registration page.
"""
response = self.client.get(
reverse('register_user'), follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL))
self.assertEquals(('/dashboard', 302),
response.redirect_chain[-1])
self.assertIn(SESSION_KEY, self.client.session)
@skip_unless_cms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_cms_registration_page_bypass(self):
"""
This tests to make sure when immediate signup is on that
the user doesn't get presented with the registration page.
"""
response = self.client.get(
reverse('signup'), follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL)
)
self.assertEqual(response.status_code, 404)
# assert that we are logged in
self.assertIn(SESSION_KEY, self.client.session)
# Now that we are logged in, make sure we don't see the registration page
response = self.client.get(reverse('signup'), follow=True)
self.assertEqual(response.status_code, 404)
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_signin_page_bypass(self):
"""
This tests to make sure when ssl authentication is on
that user doesn't get presented with the login page if they
have a certificate.
"""
# Test that they do signin if they don't have a cert
response = self.client.get(reverse('signin_user'))
self.assertEqual(200, response.status_code)
self.assertIn('login-and-registration-container', response.content)
# And get directly logged in otherwise
response = self.client.get(
reverse('signin_user'), follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL))
self.assertEquals(('/dashboard', 302),
response.redirect_chain[-1])
self.assertIn(SESSION_KEY, self.client.session)
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_ssl_bad_eamap(self):
"""
This tests the response when a user exists but their eamap
password doesn't match their internal password.
The internal password use for certificates has been removed
and this should not fail.
"""
# Create account, break internal password, and activate account
with self._create_ssl_request('/') as request:
external_auth_views.ssl_login(request)
user = User.objects.get(email=self.USER_EMAIL)
user.set_password('not autogenerated')
user.is_active = True
user.save()
# Make sure we can still login
self.client.get(
reverse('signin_user'), follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL))
self.assertIn(SESSION_KEY, self.client.session)
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITHOUT_SSL_AUTH)
def test_ssl_decorator_no_certs(self):
"""Make sure no external auth happens without SSL enabled"""
dec_mock = external_auth_views.ssl_login_shortcut(self.mock)
with self._create_normal_request(self.MOCK_URL) as request:
request.user = AnonymousUser()
# Call decorated mock function to make sure it passes
# the call through without hitting the external_auth functions and
# thereby creating an external auth map object.
dec_mock(request)
self.assertTrue(self.mock.called)
self.assertEqual(0, len(ExternalAuthMap.objects.all()))
@skip_unless_lms
def test_ssl_login_decorator(self):
"""Create mock function to test ssl login decorator"""
dec_mock = external_auth_views.ssl_login_shortcut(self.mock)
# Test that anonymous without cert doesn't create authmap
with self._create_normal_request(self.MOCK_URL) as request:
dec_mock(request)
self.assertTrue(self.mock.called)
self.assertEqual(0, len(ExternalAuthMap.objects.all()))
# Test valid user
self.mock.reset_mock()
with self._create_ssl_request(self.MOCK_URL) as request:
dec_mock(request)
self.assertFalse(self.mock.called)
self.assertEqual(1, len(ExternalAuthMap.objects.all()))
# Test logged in user gets called
self.mock.reset_mock()
with self._create_ssl_request(self.MOCK_URL) as request:
request.user = UserFactory()
dec_mock(request)
self.assertTrue(self.mock.called)
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_IMMEDIATE_SIGNUP)
def test_ssl_decorator_auto_signup(self):
"""
Test that with auto signup the decorator
will bypass registration and call retfun.
"""
dec_mock = external_auth_views.ssl_login_shortcut(self.mock)
with self._create_ssl_request(self.MOCK_URL) as request:
dec_mock(request)
# Assert our user exists in both eamap and Users
try:
ExternalAuthMap.objects.get(external_id=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to external auth map, exception was {0}'.format(str(ex)))
try:
User.objects.get(email=self.USER_EMAIL)
except ExternalAuthMap.DoesNotExist as ex:
self.fail(u'User did not get properly added to internal users, exception was {0}'.format(str(ex)))
self.assertEqual(1, len(ExternalAuthMap.objects.all()))
self.assertTrue(self.mock.called)
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_AUTO_ACTIVATE)
def test_ssl_lms_redirection(self):
"""
Auto signup auth user and ensure they return to the original
url they visited after being logged in.
"""
course = CourseFactory.create(
org='MITx',
number='999',
display_name='Robot Super Course'
)
with self._create_ssl_request('/') as request:
external_auth_views.ssl_login(request)
user = User.objects.get(email=self.USER_EMAIL)
CourseEnrollment.enroll(user, course.id)
course_private_url = '/courses/MITx/999/Robot_Super_Course/courseware'
self.assertNotIn(SESSION_KEY, self.client.session)
response = self.client.get(
course_private_url,
follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL),
HTTP_ACCEPT='text/html'
)
self.assertEqual((course_private_url, 302),
response.redirect_chain[-1])
self.assertIn(SESSION_KEY, self.client.session)
@skip_unless_cms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_AUTO_ACTIVATE)
def test_ssl_cms_redirection(self):
"""
Auto signup auth user and ensure they return to the original
url they visited after being logged in.
"""
course = CourseFactory.create(
org='MITx',
number='999',
display_name='Robot Super Course'
)
with self._create_ssl_request('/') as request:
external_auth_views.ssl_login(request)
user = User.objects.get(email=self.USER_EMAIL)
CourseEnrollment.enroll(user, course.id)
CourseStaffRole(course.id).add_users(user)
course_private_url = reverse('course_handler', args=(text_type(course.id),))
self.assertNotIn(SESSION_KEY, self.client.session)
response = self.client.get(
course_private_url,
follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL),
HTTP_ACCEPT='text/html'
)
self.assertEqual((course_private_url, 302),
response.redirect_chain[-1])
self.assertIn(SESSION_KEY, self.client.session)
@skip("This is causing tests to fail for DOP deprecation. Skip this test"
"because we are deprecating external_auth anyway (See DEPR-6 for more info).")
@skip_unless_lms
@override_settings(FEATURES=FEATURES_WITH_SSL_AUTH_AUTO_ACTIVATE)
def test_ssl_logout(self):
"""
Because the branding view is cached for anonymous users and we
use that to login users, the browser wasn't actually making the
request to that view as the redirect was being cached. This caused
a redirect loop, and this test confirms that that won't happen.
Test is only in LMS because we don't use / in studio to login SSL users.
"""
response = self.client.get(
reverse('dashboard'), follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL))
self.assertEquals(('/dashboard', 302),
response.redirect_chain[-1])
self.assertIn(SESSION_KEY, self.client.session)
response = self.client.get(
reverse('logout'), follow=True,
SSL_CLIENT_S_DN=self.AUTH_DN.format(self.USER_NAME, self.USER_EMAIL)
)
# Make sure that even though we logged out, we have logged back in
self.assertIn(SESSION_KEY, self.client.session)

962
openedx/core/djangoapps/external_auth/views.py
View File

@@ -1,962 +0,0 @@
"""
External Auth Views
"""
import fnmatch
import functools
import json
import logging
import re
import unicodedata
import urllib
from textwrap import dedent
import django_openid_auth.views as openid_views
from django.conf import settings
from django.contrib.auth import REDIRECT_FIELD_NAME, authenticate, login
from django.contrib.auth.models import User
from django.core.exceptions import ValidationError
from django.urls import reverse
from django.core.validators import validate_email
from django.db import transaction
from django.http import HttpResponse, HttpResponseForbidden, HttpResponseRedirect
from django.shortcuts import redirect
from django.utils.http import is_safe_url, urlquote
from django.utils.translation import ugettext as _
from django.views.decorators.csrf import csrf_exempt, ensure_csrf_cookie
from django_openid_auth import auth as openid_auth
from opaque_keys.edx.keys import CourseKey
from openid.consumer.consumer import SUCCESS
from openid.extensions import ax, sreg
from openid.server.server import ProtocolError, Server, UntrustedReturnURL
from openid.server.trustroot import TrustRoot
from ratelimitbackend.exceptions import RateLimitException
import student.views
from edxmako.shortcuts import render_to_response, render_to_string
from openedx.core.djangoapps.external_auth.djangostore import DjangoOpenIDStore
from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
from openedx.core.djangoapps.site_configuration.helpers import get_value
from openedx.core.djangoapps.user_api.accounts.utils import generate_password
from student.helpers import get_next_url_for_login_page
from student.models import UserProfile
from util.db import outer_atomic
from xmodule.modulestore.django import modulestore
if settings.FEATURES.get('AUTH_USE_CAS'):
from django_cas.views import login as django_cas_login
log = logging.getLogger("edx.external_auth")
AUDIT_LOG = logging.getLogger("audit")
SHIBBOLETH_DOMAIN_PREFIX = settings.SHIBBOLETH_DOMAIN_PREFIX
OPENID_DOMAIN_PREFIX = settings.OPENID_DOMAIN_PREFIX
# -----------------------------------------------------------------------------
# OpenID Common
# -----------------------------------------------------------------------------
@csrf_exempt
def default_render_failure(request, # pylint: disable=unused-argument
message,
status=403,
template_name='extauth_failure.html',
exception=None):
"""Render an Openid error page to the user"""
log.debug("In openid_failure " + message)
data = render_to_string(template_name,
dict(message=message, exception=exception))
return HttpResponse(data, status=status)
# -----------------------------------------------------------------------------
# OpenID Authentication
# -----------------------------------------------------------------------------
@transaction.non_atomic_requests
@csrf_exempt
def openid_login_complete(request,
redirect_field_name=REDIRECT_FIELD_NAME, # pylint: disable=unused-argument
render_failure=None):
"""
Complete the openid login process
"""
render_failure = (render_failure or default_render_failure)
openid_response = openid_views.parse_openid_response(request)
if not openid_response:
return render_failure(request,
'This is an OpenID relying party endpoint.')
if openid_response.status == SUCCESS:
external_id = openid_response.identity_url
oid_backend = openid_auth.OpenIDBackend()
details = oid_backend._extract_user_details(openid_response) # pylint: disable=protected-access
log.debug(u'openid success, details=%s', details)
url = getattr(settings, 'OPENID_SSO_SERVER_URL', None)
external_domain = u"{0}{1}".format(OPENID_DOMAIN_PREFIX, url)
fullname = u'%s %s' % (details.get('first_name', ''),
details.get('last_name', ''))
return _external_login_or_signup(
request,
external_id,
external_domain,
details,
details.get('email', ''),
fullname,
retfun=functools.partial(redirect, get_next_url_for_login_page(request)),
)
return render_failure(request, 'Openid failure')
def _external_login_or_signup(request,
external_id,
external_domain,
credentials,
email,
fullname,
retfun=None):
"""
Generic external auth login or signup
"""
# pylint: disable=too-many-statements
# see if we have a map from this external_id to an edX username
eamap_defaults = {
'external_credentials': json.dumps(credentials),
'external_email': email,
'external_name': fullname,
'internal_password': generate_password()
}
# We are not guaranteed to be in a transaction here since some upstream views
# use non_atomic_requests
with outer_atomic():
eamap, created = ExternalAuthMap.objects.get_or_create(
external_id=external_id,
external_domain=external_domain,
defaults=eamap_defaults
)
if created:
log.debug(u'Created eamap=%s', eamap)
else:
log.debug(u'Found eamap=%s', eamap)
log.info(u"External_Auth login_or_signup for %s : %s : %s : %s", external_domain, external_id, email, fullname)
uses_shibboleth = settings.FEATURES.get('AUTH_USE_SHIB') and external_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX)
uses_certs = settings.FEATURES.get('AUTH_USE_CERTIFICATES')
internal_user = eamap.user
if internal_user is None:
if uses_shibboleth:
# If we are using shib, try to link accounts
# For Stanford shib, the email the idp returns is actually under the control of the user.
# Since the id the idps return is not user-editable, and is of the from "username@stanford.edu",
# use the id to link accounts instead.
try:
with outer_atomic():
link_user = User.objects.get(email=eamap.external_id)
if not ExternalAuthMap.objects.filter(user=link_user).exists():
# if there's no pre-existing linked eamap, we link the user
eamap.user = link_user
eamap.save()
internal_user = link_user
log.info(u'SHIB: Linking existing account for %s', eamap.external_id)
# now pass through to log in
else:
# otherwise, there must have been an error, b/c we've already linked a user with these external
# creds
failure_msg = _(
u"You have already created an account using "
u"an external login like WebAuth or Shibboleth. "
"Please contact {tech_support_email} for support." # pylint: disable=unicode-format-string
).format(
tech_support_email=get_value('email_from_address', settings.TECH_SUPPORT_EMAIL),
)
return default_render_failure(request, failure_msg)
except User.DoesNotExist:
log.info(u'SHIB: No user for %s yet, doing signup', eamap.external_email)
return _signup(request, eamap, retfun)
else:
log.info(u'No user for %s yet. doing signup', eamap.external_email)
return _signup(request, eamap, retfun)
# We trust shib's authentication, so no need to authenticate using the password again
uname = internal_user.username
if uses_shibboleth:
user = internal_user
# Assuming this 'AUTHENTICATION_BACKENDS' is set in settings, which I think is safe
if settings.AUTHENTICATION_BACKENDS:
auth_backend = settings.AUTHENTICATION_BACKENDS[0]
else:
auth_backend = 'ratelimitbackend.backends.RateLimitModelBackend'
user.backend = auth_backend
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.info(u'Linked user.id: {0} logged in via Shibboleth'.format(user.id))
else:
AUDIT_LOG.info(u'Linked user "{0}" logged in via Shibboleth'.format(user.email))
elif uses_certs:
# Certificates are trusted, so just link the user and log the action
user = internal_user
user.backend = 'ratelimitbackend.backends.RateLimitModelBackend'
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.info(u'Linked user_id {0} logged in via SSL certificate'.format(user.id))
else:
AUDIT_LOG.info(u'Linked user "{0}" logged in via SSL certificate'.format(user.email))
else:
user = authenticate(username=uname, password=eamap.internal_password, request=request)
if user is None:
# we want to log the failure, but don't want to log the password attempted:
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.warning(u'External Auth Login failed')
else:
AUDIT_LOG.warning(u'External Auth Login failed for "{0}"'.format(uname))
return _signup(request, eamap, retfun)
if not user.is_active:
if settings.FEATURES.get('BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH'):
# if BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH, we trust external auth and activate any users
# that aren't already active
user.is_active = True
user.save()
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.info(u'Activating user {0} due to external auth'.format(user.id))
else:
AUDIT_LOG.info(u'Activating user "{0}" due to external auth'.format(uname))
else:
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.warning(u'User {0} is not active after external login'.format(user.id))
else:
AUDIT_LOG.warning(u'User "{0}" is not active after external login'.format(uname))
# TODO: improve error page
msg = 'Account not yet activated: please look for link in your email'
return default_render_failure(request, msg)
login(request, user)
request.session.set_expiry(0)
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.info(u"Login success - user.id: {0}".format(user.id))
else:
AUDIT_LOG.info(u"Login success - {0} ({1})".format(user.username, user.email))
if retfun is None:
return redirect('/')
return retfun()
def _flatten_to_ascii(txt):
"""
Flattens possibly unicode txt to ascii (django username limitation)
@param name:
@return: the flattened txt (in the same type as was originally passed in)
"""
if isinstance(txt, str):
txt = txt.decode('utf-8')
return unicodedata.normalize('NFKD', txt).encode('ASCII', 'ignore')
else:
return unicode(unicodedata.normalize('NFKD', txt).encode('ASCII', 'ignore'))
@ensure_csrf_cookie
def _signup(request, eamap, retfun=None):
"""
Present form to complete for signup via external authentication.
Even though the user has external credentials, he/she still needs
to create an account on the edX system, and fill in the user
registration form.
eamap is an ExternalAuthMap object, specifying the external user
for which to complete the signup.
retfun is a function to execute for the return value, if immediate
signup is used. That allows @ssl_login_shortcut() to work.
"""
from openedx.core.djangoapps.user_authn.views.deprecated import create_account, register_user
# save this for use by create_account
request.session['ExternalAuthMap'] = eamap
if settings.FEATURES.get('AUTH_USE_CERTIFICATES_IMMEDIATE_SIGNUP', ''):
# do signin immediately, by calling create_account, instead of asking
# student to fill in form. MIT students already have information filed.
username = eamap.external_email.split('@', 1)[0]
username = username.replace('.', '_')
post_vars = dict(username=username,
honor_code=u'true',
terms_of_service=u'true')
log.info(u'doing immediate signup for %s, params=%s', username, post_vars)
create_account(request, post_vars)
# should check return content for successful completion before
if retfun is not None:
return retfun()
else:
return redirect('/')
# default conjoin name, no spaces, flattened to ascii b/c django can't handle unicode usernames, sadly
# but this only affects username, not fullname
username = re.sub(r'\s', '', _flatten_to_ascii(eamap.external_name), flags=re.UNICODE)
context = {
'has_extauth_info': True,
'show_signup_immediately': True,
'extauth_domain': eamap.external_domain,
'extauth_id': eamap.external_id,
'extauth_email': eamap.external_email,
'extauth_username': username,
'extauth_name': eamap.external_name,
'ask_for_tos': True,
}
# Some openEdX instances can't have terms of service for shib users, like
# according to Stanford's Office of General Counsel
uses_shibboleth = (settings.FEATURES.get('AUTH_USE_SHIB') and
eamap.external_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX))
if uses_shibboleth and settings.FEATURES.get('SHIB_DISABLE_TOS'):
context['ask_for_tos'] = False
# detect if full name is blank and ask for it from user
context['ask_for_fullname'] = eamap.external_name.strip() == ''
# validate provided mail and if it's not valid ask the user
try:
validate_email(eamap.external_email)
context['ask_for_email'] = False
except ValidationError:
context['ask_for_email'] = True
log.info(u'EXTAUTH: Doing signup for %s', eamap.external_id)
return register_user(request, extra_context=context)
# -----------------------------------------------------------------------------
# MIT SSL
# -----------------------------------------------------------------------------
def _ssl_dn_extract_info(dn_string):
"""
Extract username, email address (may be anyuser@anydomain.com) and
full name from the SSL DN string. Return (user,email,fullname) if
successful, and None otherwise.
"""
search_string = re.search('/emailAddress=(.*)@([^/]+)', dn_string)
if search_string:
user = search_string.group(1)
email = "%s@%s" % (user, search_string.group(2))
else:
raise ValueError
search_string = re.search('/CN=([^/]+)/', dn_string)
if search_string:
fullname = search_string.group(1)
else:
raise ValueError
return (user, email, fullname)
def ssl_get_cert_from_request(request):
"""
Extract user information from certificate, if it exists, returning (user, email, fullname).
Else return None.
"""
certkey = "SSL_CLIENT_S_DN" # specify the request.META field to use
cert = request.META.get(certkey, '')
if not cert:
cert = request.META.get('HTTP_' + certkey, '')
if not cert:
try:
# try the direct apache2 SSL key
cert = request._req.subprocess_env.get(certkey, '') # pylint: disable=protected-access
except Exception: # pylint: disable=broad-except
return ''
return cert
def ssl_login_shortcut(func):
"""
Python function decorator for login procedures, to allow direct login
based on existing ExternalAuth record and MIT ssl certificate.
"""
@transaction.non_atomic_requests
def wrapped(*args, **kwargs):
"""
This manages the function wrapping, by determining whether to inject
the _external signup or just continuing to the internal function
call.
"""
if not settings.FEATURES['AUTH_USE_CERTIFICATES']:
return func(*args, **kwargs)
request = args[0]
if request.user and request.user.is_authenticated: # don't re-authenticate
return func(*args, **kwargs)
cert = ssl_get_cert_from_request(request)
if not cert: # no certificate information - show normal login window
return func(*args, **kwargs)
def retfun():
"""Wrap function again for call by _external_login_or_signup"""
return func(*args, **kwargs)
(_user, email, fullname) = _ssl_dn_extract_info(cert)
return _external_login_or_signup(
request,
external_id=email,
external_domain="ssl:MIT",
credentials=cert,
email=email,
fullname=fullname,
retfun=retfun
)
return wrapped
@csrf_exempt
def ssl_login(request):
"""
This is called by branding.views.index when
FEATURES['AUTH_USE_CERTIFICATES'] = True
Used for MIT user authentication. This presumes the web server
(nginx) has been configured to require specific client
certificates.
If the incoming protocol is HTTPS (SSL) then authenticate via
client certificate. The certificate provides user email and
fullname; this populates the ExternalAuthMap. The user is
nevertheless still asked to complete the edX signup.
Else continues on with student.views.index, and no authentication.
"""
# Just to make sure we're calling this only at MIT:
if not settings.FEATURES['AUTH_USE_CERTIFICATES']:
return HttpResponseForbidden()
cert = ssl_get_cert_from_request(request)
if not cert:
# no certificate information - go onward to main index
return student.views.index(request)
(_user, email, fullname) = _ssl_dn_extract_info(cert)
redirect_to = get_next_url_for_login_page(request)
retfun = functools.partial(redirect, redirect_to)
return _external_login_or_signup(
request,
external_id=email,
external_domain="ssl:MIT",
credentials=cert,
email=email,
fullname=fullname,
retfun=retfun
)
# -----------------------------------------------------------------------------
# CAS (Central Authentication Service)
# -----------------------------------------------------------------------------
def cas_login(request, next_page=None, required=False):
"""
Uses django_cas for authentication.
CAS is a common authentcation method pioneered by Yale.
See http://en.wikipedia.org/wiki/Central_Authentication_Service
Does normal CAS login then generates user_profile if nonexistent,
and if login was successful. We assume that user details are
maintained by the central service, and thus an empty user profile
is appropriate.
"""
ret = django_cas_login(request, next_page, required)
if request.user.is_authenticated:
user = request.user
UserProfile.objects.get_or_create(
user=user,
defaults={'name': user.username}
)
return ret
# -----------------------------------------------------------------------------
# Shibboleth (Stanford and others. Uses *Apache* environment variables)
# -----------------------------------------------------------------------------
@transaction.non_atomic_requests
def shib_login(request):
"""
Uses Apache's REMOTE_USER environment variable as the external id.
This in turn typically uses EduPersonPrincipalName
http://www.incommonfederation.org/attributesummary.html#eduPersonPrincipal
but the configuration is in the shibboleth software.
"""
shib_error_msg = dedent(_(
"""
Your university identity server did not return your ID information to us.
Please try logging in again. (You may need to restart your browser.)
"""))
if not request.META.get('REMOTE_USER'):
log.error(u"SHIB: no REMOTE_USER found in request.META")
return default_render_failure(request, shib_error_msg)
elif not request.META.get('Shib-Identity-Provider'):
log.error(u"SHIB: no Shib-Identity-Provider in request.META")
return default_render_failure(request, shib_error_msg)
else:
# If we get here, the user has authenticated properly
shib = {attr: request.META.get(attr, '').decode('utf-8')
for attr in ['REMOTE_USER', 'givenName', 'sn', 'mail', 'Shib-Identity-Provider', 'displayName']}
# Clean up first name, last name, and email address
# TODO: Make this less hardcoded re: format, but split will work
# even if ";" is not present, since we are accessing 1st element
shib['sn'] = shib['sn'].split(";")[0].strip().capitalize()
shib['givenName'] = shib['givenName'].split(";")[0].strip().capitalize()
# TODO: should we be logging creds here, at info level?
log.info(u"SHIB creds returned: %r", shib)
fullname = shib['displayName'] if shib['displayName'] else u'%s %s' % (shib['givenName'], shib['sn'])
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,
external_id=shib['REMOTE_USER'],
external_domain=SHIBBOLETH_DOMAIN_PREFIX + shib['Shib-Identity-Provider'],
credentials=shib,
email=shib['mail'],
fullname=fullname,
retfun=retfun
)
def _safe_postlogin_redirect(redirect_to, safehost, default_redirect='/'):
"""
If redirect_to param is safe (not off this host), then perform the redirect.
Otherwise just redirect to '/'.
Basically copied from django.contrib.auth.views.login
@param redirect_to: user-supplied redirect url
@param safehost: which host is safe to redirect to
@return: an HttpResponseRedirect
"""
if is_safe_url(url=redirect_to, allowed_hosts={safehost}, require_https=True):
return redirect(redirect_to)
return redirect(default_redirect)
def course_specific_login(request, course_id):
"""
Dispatcher function for selecting the specific login method
required by the course
"""
course_key = CourseKey.from_string(course_id)
course = modulestore().get_course(course_key)
if not course:
# couldn't find the course, will just return vanilla signin page
return redirect_with_get('signin_user', request.GET)
# now the dispatching conditionals. Only shib for now
if (
settings.FEATURES.get('AUTH_USE_SHIB') and
course.enrollment_domain and
course.enrollment_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX)
):
return redirect_with_get('shib-login', request.GET)
# Default fallthrough to normal signin page
return redirect_with_get('signin_user', request.GET)
def course_specific_register(request, course_id):
"""
Dispatcher function for selecting the specific registration method
required by the course
"""
course_key = CourseKey.from_string(course_id)
course = modulestore().get_course(course_key)
if not course:
# couldn't find the course, will just return vanilla registration page
return redirect_with_get('register_user', request.GET)
# now the dispatching conditionals. Only shib for now
if (
settings.FEATURES.get('AUTH_USE_SHIB') and
course.enrollment_domain and
course.enrollment_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX)
):
# shib-login takes care of both registration and login flows
return redirect_with_get('shib-login', request.GET)
# Default fallthrough to normal registration page
return redirect_with_get('register_user', request.GET)
def redirect_with_get(view_name, get_querydict, do_reverse=True):
"""
Helper function to carry over get parameters across redirects
Using urlencode(safe='/') because the @login_required decorator generates 'next' queryparams with '/' unencoded
"""
if do_reverse:
url = reverse(view_name)
else:
url = view_name
if get_querydict:
return redirect("%s?%s" % (url, get_querydict.urlencode(safe='/')))
return redirect(view_name)
# -----------------------------------------------------------------------------
# OpenID Provider
# -----------------------------------------------------------------------------
def get_xrds_url(resource, request):
"""
Return the XRDS url for a resource
"""
host = request.get_host()
location = host + '/openid/provider/' + resource + '/'
if request.is_secure():
return 'https://' + location
else:
return 'http://' + location
def add_openid_simple_registration(request, response, data):
"""
Add simple registration fields to the response if requested.
"""
sreg_data = {}
sreg_request = sreg.SRegRequest.fromOpenIDRequest(request)
sreg_fields = sreg_request.allRequestedFields()
# if consumer requested simple registration fields, add them
if sreg_fields:
for field in sreg_fields:
if field == 'email' and 'email' in data:
sreg_data['email'] = data['email']
elif field == 'fullname' and 'fullname' in data:
sreg_data['fullname'] = data['fullname']
elif field == 'nickname' and 'nickname' in data:
sreg_data['nickname'] = data['nickname']
# construct sreg response
sreg_response = sreg.SRegResponse.extractResponse(sreg_request,
sreg_data)
sreg_response.toMessage(response.fields)
def add_openid_attribute_exchange(request, response, data):
"""
Add attribute exchange fields to the response if requested.
"""
try:
ax_request = ax.FetchRequest.fromOpenIDRequest(request)
except ax.AXError:
# not using OpenID attribute exchange extension
pass
else:
ax_response = ax.FetchResponse()
# if consumer requested attribute exchange fields, add them
if ax_request and ax_request.requested_attributes:
for type_uri in ax_request.requested_attributes.iterkeys():
email_schema = 'http://axschema.org/contact/email'
name_schema = 'http://axschema.org/namePerson'
if type_uri == email_schema and 'email' in data:
ax_response.addValue(email_schema, data['email'])
elif type_uri == name_schema and 'fullname' in data:
ax_response.addValue(name_schema, data['fullname'])
# construct ax response
ax_response.toMessage(response.fields)
def provider_respond(server, request, response, data):
"""
Respond to an OpenID request
"""
# get and add extensions
add_openid_simple_registration(request, response, data)
add_openid_attribute_exchange(request, response, data)
# create http response from OpenID response
webresponse = server.encodeResponse(response)
http_response = HttpResponse(webresponse.body)
http_response.status_code = webresponse.code
# add OpenID headers to response
for key, val in webresponse.headers.iteritems():
http_response[key] = val
return http_response
def validate_trust_root(openid_request):
"""
Only allow OpenID requests from valid trust roots
"""
trusted_roots = getattr(settings, 'OPENID_PROVIDER_TRUSTED_ROOT', None)
if not trusted_roots:
# not using trusted roots
return True
# don't allow empty trust roots
if (not hasattr(openid_request, 'trust_root') or
not openid_request.trust_root):
log.error('no trust_root')
return False
# ensure trust root parses cleanly (one wildcard, of form *.foo.com, etc.)
trust_root = TrustRoot.parse(openid_request.trust_root)
if not trust_root:
log.error('invalid trust_root')
return False
# don't allow empty return tos
if (not hasattr(openid_request, 'return_to') or
not openid_request.return_to):
log.error('empty return_to')
return False
# ensure return to is within trust root
if not trust_root.validateURL(openid_request.return_to):
log.error('invalid return_to')
return False
# check that the root matches the ones we trust
if not any(r for r in trusted_roots if fnmatch.fnmatch(trust_root, r)):
log.error('non-trusted root')
return False
return True
@csrf_exempt
def provider_login(request):
"""
OpenID login endpoint
"""
# pylint: disable=too-many-statements
# make and validate endpoint
endpoint = get_xrds_url('login', request)
if not endpoint:
return default_render_failure(request, "Invalid OpenID request")
# initialize store and server
store = DjangoOpenIDStore()
server = Server(store, endpoint)
# first check to see if the request is an OpenID request.
# If so, the client will have specified an 'openid.mode' as part
# of the request.
if request.method == 'GET':
querydict = dict(request.GET.items())
else:
querydict = dict(request.POST.items())
error = False
if 'openid.mode' in request.GET or 'openid.mode' in request.POST:
# decode request
try:
openid_request = server.decodeRequest(querydict)
except (UntrustedReturnURL, ProtocolError):
openid_request = None
if not openid_request:
return default_render_failure(request, "Invalid OpenID request")
# don't allow invalid and non-trusted trust roots
if not validate_trust_root(openid_request):
return default_render_failure(request, "Invalid OpenID trust root")
# checkid_immediate not supported, require user interaction
if openid_request.mode == 'checkid_immediate':
return provider_respond(server, openid_request,
openid_request.answer(False), {})
# checkid_setup, so display login page
# (by falling through to the provider_login at the
# bottom of this method).
elif openid_request.mode == 'checkid_setup':
if openid_request.idSelect():
# remember request and original path
request.session['openid_setup'] = {
'request': openid_request,
'url': request.get_full_path(),
'post_params': request.POST,
}
# user failed login on previous attempt
if 'openid_error' in request.session:
error = True
del request.session['openid_error']
# OpenID response
else:
return provider_respond(server, openid_request,
server.handleRequest(openid_request), {})
# handle login redirection: these are also sent to this view function,
# but are distinguished by lacking the openid mode. We also know that
# they are posts, because they come from the popup
elif request.method == 'POST' and 'openid_setup' in request.session:
# get OpenID request from session
openid_setup = request.session['openid_setup']
openid_request = openid_setup['request']
openid_request_url = openid_setup['url']
post_params = openid_setup['post_params']
# We need to preserve the parameters, and the easiest way to do this is
# through the URL
url_post_params = {
param: post_params[param] for param in post_params if param.startswith('openid')
}
encoded_params = urllib.urlencode(url_post_params)
if '?' not in openid_request_url:
openid_request_url = openid_request_url + '?' + encoded_params
else:
openid_request_url = openid_request_url + '&' + encoded_params
del request.session['openid_setup']
# don't allow invalid trust roots
if not validate_trust_root(openid_request):
return default_render_failure(request, "Invalid OpenID trust root")
# check if user with given email exists
# Failure is redirected to this method (by using the original URL),
# which will bring up the login dialog.
email = request.POST.get('email', None)
try:
user = User.objects.get(email=email)
except User.DoesNotExist:
request.session['openid_error'] = True
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.warning(u"OpenID login failed - Unknown user email")
else:
msg = u"OpenID login failed - Unknown user email: {0}".format(email)
AUDIT_LOG.warning(msg)
return HttpResponseRedirect(openid_request_url)
# attempt to authenticate user (but not actually log them in...)
# Failure is again redirected to the login dialog.
username = user.username
password = request.POST.get('password', None)
try:
user = authenticate(username=username, password=password, request=request)
except RateLimitException:
AUDIT_LOG.warning(u'OpenID - Too many failed login attempts.')
return HttpResponseRedirect(openid_request_url)
if user is None:
request.session['openid_error'] = True
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.warning(u"OpenID login failed - invalid password")
else:
AUDIT_LOG.warning(
u"OpenID login failed - password for %s is invalid", email)
return HttpResponseRedirect(openid_request_url)
# authentication succeeded, so fetch user information
# that was requested
if user is not None and user.is_active:
# remove error from session since login succeeded
if 'openid_error' in request.session:
del request.session['openid_error']
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.info(u"OpenID login success - user.id: %s", user.id)
else:
AUDIT_LOG.info(
u"OpenID login success - %s (%s)", user.username, user.email)
# redirect user to return_to location
url = endpoint + urlquote(user.username)
response = openid_request.answer(True, None, url)
# Note too that this is hardcoded, and not really responding to
# the extensions that were registered in the first place.
results = {
'nickname': user.username,
'email': user.email,
'fullname': user.profile.name,
}
# the request succeeded:
return provider_respond(server, openid_request, response, results)
# the account is not active, so redirect back to the login page:
request.session['openid_error'] = True
if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
AUDIT_LOG.warning(
u"Login failed - Account not active for user.id %s", user.id)
else:
AUDIT_LOG.warning(
u"Login failed - Account not active for user %s", username)
return HttpResponseRedirect(openid_request_url)
# determine consumer domain if applicable
return_to = request.GET.get('openid.return_to') or request.POST.get('openid.return_to') or ''
if return_to:
matches = re.match(r'\w+:\/\/([\w\.-]+)', return_to)
return_to = matches.group(1)
# display login page
response = render_to_response('provider_login.html', {
'error': error,
'return_to': return_to
})
# add custom XRDS header necessary for discovery process
response['X-XRDS-Location'] = get_xrds_url('xrds', request)
return response
def provider_identity(request):
"""
XRDS for identity discovery
"""
response = render_to_response('identity.xml',
{'url': get_xrds_url('login', request)},
content_type='text/xml')
# custom XRDS header necessary for discovery process
response['X-XRDS-Location'] = get_xrds_url('identity', request)
return response
def provider_xrds(request):
"""
XRDS for endpoint discovery
"""
response = render_to_response('xrds.xml',
{'url': get_xrds_url('login', request)},
content_type='text/xml')
# custom XRDS header necessary for discovery process
response['X-XRDS-Location'] = get_xrds_url('xrds', request)
return response

9
openedx/core/djangoapps/user_authn/views/deprecated.py
View File

@@ -15,8 +15,6 @@ from edxmako.shortcuts import render_to_response
from openedx.core.djangoapps.user_authn.views.register import create_account_with_params
from openedx.core.djangoapps.user_authn.cookies import set_logged_in_cookies
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.site_configuration import helpers as configuration_helpers
from openedx.core.djangoapps.user_api.config.waffle import PREVENT_AUTH_USER_WRITES, SYSTEM_MAINTENANCE_MSG, waffle
from student.helpers import (
@@ -32,9 +30,6 @@ from util.json_request import JsonResponse
@ensure_csrf_cookie
def signin_user(request):
"""Deprecated. To be replaced by :class:`user_authn.views.login_form.login_and_registration_form`."""
external_auth_response = external_auth_login(request)
if external_auth_response is not None:
return external_auth_response
# Determine the URL to redirect to following login:
redirect_to = get_next_url_for_login_page(request)
if request.user.is_authenticated:
@@ -74,10 +69,6 @@ def register_user(request, extra_context=None):
if request.user.is_authenticated:
return redirect(redirect_to)
external_auth_response = external_auth_register(request)
if external_auth_response is not None:
return external_auth_response
context = {
'login_redirect_url': redirect_to, # This gets added to the query string of the "Sign In" button in the header
'email': '',

19
openedx/core/djangoapps/user_authn/views/login.py
View File

@@ -20,8 +20,6 @@ from ratelimitbackend.exceptions import RateLimitException
from edxmako.shortcuts import render_to_response
from openedx.core.djangoapps.user_authn.cookies import set_logged_in_cookies, refresh_jwt_cookies
from openedx.core.djangoapps.user_authn.exceptions import AuthFailedError
import openedx.core.djangoapps.external_auth.views
from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
from openedx.core.djangoapps.password_policy import compliance as password_policy_compliance
from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
from openedx.core.djangoapps.util.user_messages import PageLevelMessages
@@ -105,22 +103,6 @@ def _get_user_by_email(request):
AUDIT_LOG.warning(u"Login failed - Unknown user email: {0}".format(email))
def _check_shib_redirect(user):
"""
See if the user has a linked shibboleth account, if so, redirect the user to shib-login.
This behavior is pretty much like what gmail does for shibboleth. Try entering some @stanford.edu
address into the Gmail login.
"""
if settings.FEATURES.get('AUTH_USE_SHIB') and user:
try:
eamap = ExternalAuthMap.objects.get(user=user)
if eamap.external_domain.startswith(openedx.core.djangoapps.external_auth.views.SHIBBOLETH_DOMAIN_PREFIX):
raise AuthFailedError('', redirect=reverse('shib-login'))
except ExternalAuthMap.DoesNotExist:
# This is actually the common case, logging in user without external linked login
AUDIT_LOG.info(u"User %s w/o external auth attempting login", user)
def _check_excessive_login_attempts(user):
"""
See if account has been locked out due to excessive login failures
@@ -347,7 +329,6 @@ def login_user(request):
else:
email_user = _get_user_by_email(request)
_check_shib_redirect(email_user)
_check_excessive_login_attempts(email_user)
possibly_authenticated_user = email_user

24
openedx/core/djangoapps/user_authn/views/login_form.py
View File

@@ -15,8 +15,6 @@ from edxmako.shortcuts import render_to_response
from openedx.core.djangoapps.user_authn.views.deprecated import (
register_user as old_register_view, signin_user as old_login_view
)
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.site_configuration import helpers as configuration_helpers
from openedx.core.djangoapps.theming.helpers import is_request_in_themed_site
from openedx.core.djangoapps.user_api.accounts.utils import is_secondary_email_feature_enabled
@@ -97,11 +95,6 @@ def login_and_registration_form(request, initial_mode="login"):
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 = [
{
@@ -250,20 +243,3 @@ def _third_party_auth_context(request, redirect_to, tpa_hint=None):
break
return context
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)

2
openedx/core/djangoapps/user_authn/views/logout.py
View File

@@ -24,7 +24,7 @@ class LogoutView(TemplateView):
template_name = 'logout.html'
# Keep track of the page to which the user should ultimately be redirected.
default_target = reverse_lazy('cas-logout') if settings.FEATURES.get('AUTH_USE_CAS') else '/'
default_target = '/'
def post(self, request, *args, **kwargs):
"""

62
openedx/core/djangoapps/user_authn/views/register.py
View File

@@ -132,26 +132,19 @@ def create_account_with_params(request, params):
]}
)
do_external_auth, eamap = pre_account_creation_external_auth(request, params)
extended_profile_fields = configuration_helpers.get_value('extended_profile_fields', [])
# Can't have terms of service for certain SHIB users, like at Stanford
registration_fields = getattr(settings, 'REGISTRATION_EXTRA_FIELDS', {})
tos_required = (
registration_fields.get('terms_of_service') != 'hidden' or
registration_fields.get('honor_code') != 'hidden'
) and (
not settings.FEATURES.get("AUTH_USE_SHIB") or
not settings.FEATURES.get("SHIB_DISABLE_TOS") or
not do_external_auth or
not eamap.external_domain.startswith(settings.SHIBBOLETH_DOMAIN_PREFIX)
)
form = AccountCreationForm(
data=params,
extra_fields=extra_fields,
extended_profile_fields=extended_profile_fields,
do_third_party_auth=do_external_auth,
do_third_party_auth=False,
tos_required=tos_required,
)
custom_form = get_registration_extension_form(data=params)
@@ -169,11 +162,9 @@ def create_account_with_params(request, params):
django_login(request, new_user)
request.session.set_expiry(0)
post_account_creation_external_auth(do_external_auth, eamap, new_user)
# Check if system is configured to skip activation email for the current user.
skip_email = _skip_activation_email(
user, do_external_auth, running_pipeline, third_party_provider,
user, running_pipeline, third_party_provider,
)
if skip_email:
@@ -213,51 +204,6 @@ def create_account_with_params(request, params):
return new_user
def pre_account_creation_external_auth(request, params):
"""
External auth related setup before account is created.
"""
# If doing signup for an external authorization, then get email, password, name from the eamap
# don't use the ones from the form, since the user could have hacked those
# unless originally we didn't get a valid email or name from the external auth
# TODO: We do not check whether these values meet all necessary criteria, such as email length
do_external_auth = 'ExternalAuthMap' in request.session
eamap = None
if do_external_auth:
eamap = request.session['ExternalAuthMap']
try:
validate_email(eamap.external_email)
params["email"] = eamap.external_email
except ValidationError:
pass
if len(eamap.external_name.strip()) >= accounts_settings.NAME_MIN_LENGTH:
params["name"] = eamap.external_name
params["password"] = eamap.internal_password
log.debug(u'In create_account with external_auth: user = %s, email=%s', params["name"], params["email"])
return do_external_auth, eamap
def post_account_creation_external_auth(do_external_auth, eamap, new_user):
"""
External auth related updates after account is created.
"""
if do_external_auth:
eamap.user = new_user
eamap.dtsignup = datetime.datetime.now(UTC)
eamap.save()
AUDIT_LOG.info(u"User registered with external_auth %s", new_user.username)
AUDIT_LOG.info(u'Updated ExternalAuthMap for %s to be %s', new_user.username, eamap)
if settings.FEATURES.get('BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH'):
log.info('bypassing activation email')
new_user.is_active = True
new_user.save()
AUDIT_LOG.info(
u"Login activated on extauth account - {0} ({1})".format(new_user.username, new_user.email)
)
def _link_user_to_third_party_provider(
is_third_party_auth_enabled,
third_party_auth_credentials_in_api,
@@ -352,7 +298,7 @@ def _track_user_registration(user, profile, params, third_party_provider):
)
def _skip_activation_email(user, do_external_auth, running_pipeline, third_party_provider):
def _skip_activation_email(user, running_pipeline, third_party_provider):
"""
Return `True` if activation email should be skipped.
@@ -370,7 +316,6 @@ def _skip_activation_email(user, do_external_auth, running_pipeline, third_party
Arguments:
user (User): Django User object for the current user.
do_external_auth (bool): True if external authentication is in progress.
running_pipeline (dict): Dictionary containing user and pipeline data for third party authentication.
third_party_provider (ProviderConfig): An instance of third party provider configuration.
@@ -406,7 +351,6 @@ def _skip_activation_email(user, do_external_auth, running_pipeline, third_party
return (
settings.FEATURES.get('SKIP_EMAIL_VALIDATION', None) or
settings.FEATURES.get('AUTOMATIC_AUTH_FOR_TESTING') or
(settings.FEATURES.get('BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH') and do_external_auth) or
(third_party_provider and third_party_provider.skip_email_verification and valid_email)
)

89
openedx/core/djangoapps/user_authn/views/tests/test_login.py
View File

@@ -16,9 +16,7 @@ from django.test.client import Client
from django.test.utils import override_settings
from django.urls import NoReverseMatch, reverse
from mock import patch
from six import text_type
from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
from openedx.core.djangoapps.password_policy.compliance import (
NonCompliantPasswordException,
NonCompliantPasswordWarning
@@ -28,8 +26,6 @@ from openedx.core.djangoapps.user_authn.cookies import jwt_cookies
from openedx.core.djangoapps.user_authn.tests.utils import setup_login_oauth_client
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
from student.tests.factories import RegistrationFactory, UserFactory, UserProfileFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@ddt.ddt
@@ -569,88 +565,3 @@ class LoginTest(CacheIsolationTestCase):
format_string = args[0]
for log_string in log_strings:
self.assertNotIn(log_string, format_string)
class ExternalAuthShibTest(ModuleStoreTestCase):
"""
Tests how login_user() interacts with ExternalAuth, in particular Shib
"""
def setUp(self):
super(ExternalAuthShibTest, self).setUp()
self.course = CourseFactory.create(
org='Stanford',
number='456',
display_name='NO SHIB',
user_id=self.user.id,
)
self.shib_course = CourseFactory.create(
org='Stanford',
number='123',
display_name='Shib Only',
enrollment_domain='shib:https://idp.stanford.edu/',
user_id=self.user.id,
)
self.user_w_map = UserFactory.create(email='withmap@stanford.edu')
self.extauth = ExternalAuthMap(external_id='withmap@stanford.edu',
external_email='withmap@stanford.edu',
external_domain='shib:https://idp.stanford.edu/',
external_credentials="",
user=self.user_w_map)
self.user_w_map.save()
self.extauth.save()
self.user_wo_map = UserFactory.create(email='womap@gmail.com')
self.user_wo_map.save()
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
def test_login_page_redirect(self):
"""
Tests that when a shib user types their email address into the login page, they get redirected
to the shib login.
"""
response = self.client.post(reverse('login'), {'email': self.user_w_map.email, 'password': ''})
self.assertEqual(response.status_code, 200)
obj = json.loads(response.content)
self.assertEqual(obj, {
'success': False,
'redirect': reverse('shib-login'),
})
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
def test_login_required_dashboard(self):
"""
Tests redirects to when @login_required to dashboard, which should always be the normal login,
since there is no course context
"""
response = self.client.get(reverse('dashboard'))
self.assertEqual(response.status_code, 302)
self.assertEqual(response['Location'], '/login?next=/dashboard')
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
def test_externalauth_login_required_course_context(self):
"""
Tests the redirects when visiting course-specific URL with @login_required.
Should vary by course depending on its enrollment_domain
"""
target_url = reverse('courseware', args=[text_type(self.course.id)])
noshib_response = self.client.get(target_url, follow=True, HTTP_ACCEPT="text/html")
self.assertEqual(noshib_response.redirect_chain[-1],
('/login?next={url}'.format(url=target_url), 302))
self.assertContains(noshib_response, (u"Sign in or Register | {platform_name}"
.format(platform_name=settings.PLATFORM_NAME)))
self.assertEqual(noshib_response.status_code, 200)
target_url_shib = reverse('courseware', args=[text_type(self.shib_course.id)])
shib_response = self.client.get(**{'path': target_url_shib,
'follow': True,
'REMOTE_USER': self.extauth.external_id,
'Shib-Identity-Provider': 'https://idp.stanford.edu/',
'HTTP_ACCEPT': "text/html"})
# Test that the shib-login redirect page with ?next= and the desired page are part of the redirect chain
# The 'courseware' page actually causes a redirect itself, so it's not the end of the chain and we
# won't test its contents
self.assertEqual(shib_response.redirect_chain[-3],
('/shib-login/?next={url}'.format(url=target_url_shib), 302))
self.assertEqual(shib_response.redirect_chain[-2],
(target_url_shib, 302))
self.assertEqual(shib_response.status_code, 200)

95
openedx/core/djangoapps/user_authn/views/tests/test_register.py
View File

@@ -3,7 +3,6 @@
import json
import unittest
from datetime import datetime
from importlib import import_module
import unicodedata
import ddt
@@ -19,12 +18,10 @@ from django.contrib.auth.hashers import make_password
from django_comment_common.models import ForumsConfig
from notification_prefs import NOTIFICATION_PREF_KEY
from openedx.core.djangoapps.user_authn.views.deprecated import create_account
from openedx.core.djangoapps.user_authn.views.register import (
REGISTRATION_AFFILIATE_ID, REGISTRATION_UTM_CREATED_AT, REGISTRATION_UTM_PARAMETERS,
_skip_activation_email,
)
from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY
from openedx.core.djangoapps.site_configuration.tests.mixins import SiteMixin
from openedx.core.djangoapps.user_api.accounts import (
@@ -280,63 +277,6 @@ class TestCreateAccount(SiteMixin, TestCase):
profile = self.create_account_and_fetch_profile()
self.assertIsNone(profile.year_of_birth)
def base_extauth_bypass_sending_activation_email(self, bypass_activation_email):
"""
Tests user creation without sending activation email when
doing external auth
"""
request = self.request_factory.post(self.url, self.params)
request.site = self.site
# now indicate we are doing ext_auth by setting 'ExternalAuthMap' in the session.
request.session = import_module(settings.SESSION_ENGINE).SessionStore() # empty session
extauth = ExternalAuthMap(external_id='withmap@stanford.edu',
external_email='withmap@stanford.edu',
internal_password=self.params['password'],
external_domain='shib:https://idp.stanford.edu/')
request.session['ExternalAuthMap'] = extauth
request.user = AnonymousUser()
with mock.patch('edxmako.request_context.get_current_request', return_value=request):
with mock.patch('django.core.mail.send_mail') as mock_send_mail:
create_account(request)
# check that send_mail is called
if bypass_activation_email:
self.assertFalse(mock_send_mail.called)
else:
self.assertTrue(mock_send_mail.called)
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@mock.patch.dict(settings.FEATURES,
{'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': True, 'AUTOMATIC_AUTH_FOR_TESTING': False})
def test_extauth_bypass_sending_activation_email_with_bypass(self):
"""
Tests user creation without sending activation email when
settings.FEATURES['BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH']=True and doing external auth
"""
self.base_extauth_bypass_sending_activation_email(True)
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@mock.patch.dict(settings.FEATURES,
{'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': False, 'AUTOMATIC_AUTH_FOR_TESTING': False})
def test_extauth_bypass_sending_activation_email_without_bypass_1(self):
"""
Tests user creation without sending activation email when
settings.FEATURES['BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH']=False and doing external auth
"""
self.base_extauth_bypass_sending_activation_email(False)
@unittest.skipUnless(settings.FEATURES.get('AUTH_USE_SHIB'), "AUTH_USE_SHIB not set")
@mock.patch.dict(settings.FEATURES, {'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': False,
'AUTOMATIC_AUTH_FOR_TESTING': False, 'SKIP_EMAIL_VALIDATION': True})
def test_extauth_bypass_sending_activation_email_without_bypass_2(self):
"""
Tests user creation without sending activation email when
settings.FEATURES['BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH']=False and doing external auth
"""
self.base_extauth_bypass_sending_activation_email(True)
@ddt.data(True, False)
def test_discussions_email_digest_pref(self, digest_enabled):
with mock.patch.dict("student.models.settings.FEATURES", {"ENABLE_DISCUSSION_EMAIL_DIGEST": digest_enabled}):
@@ -491,59 +431,37 @@ class TestCreateAccount(SiteMixin, TestCase):
@ddt.data(
(
False, False, get_mock_pipeline_data(),
False, get_mock_pipeline_data(),
{
'SKIP_EMAIL_VALIDATION': False, 'AUTOMATIC_AUTH_FOR_TESTING': False,
'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': False,
},
False # Do not skip activation email for normal scenario.
),
(
False, False, get_mock_pipeline_data(),
False, get_mock_pipeline_data(),
{
'SKIP_EMAIL_VALIDATION': True, 'AUTOMATIC_AUTH_FOR_TESTING': False,
'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': False,
},
True # Skip activation email when `SKIP_EMAIL_VALIDATION` FEATURE flag is active.
),
(
False, False, get_mock_pipeline_data(),
False, get_mock_pipeline_data(),
{
'SKIP_EMAIL_VALIDATION': False, 'AUTOMATIC_AUTH_FOR_TESTING': True,
'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': False,
},
True # Skip activation email when `AUTOMATIC_AUTH_FOR_TESTING` FEATURE flag is active.
),
(
True, False, get_mock_pipeline_data(),
True, get_mock_pipeline_data(),
{
'SKIP_EMAIL_VALIDATION': False, 'AUTOMATIC_AUTH_FOR_TESTING': False,
'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': True,
},
True # Skip activation email for external auth scenario.
),
(
False, False, get_mock_pipeline_data(),
{
'SKIP_EMAIL_VALIDATION': False, 'AUTOMATIC_AUTH_FOR_TESTING': False,
'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': True,
},
False # Do not skip activation email when `BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH` feature flag is set
# but it is not external auth scenario.
),
(
False, True, get_mock_pipeline_data(),
{
'SKIP_EMAIL_VALIDATION': False, 'AUTOMATIC_AUTH_FOR_TESTING': False,
'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': False,
},
True # Skip activation email if `skip_email_verification` is set for third party authentication.
),
(
False, False, get_mock_pipeline_data(email='invalid@yopmail.com'),
False, get_mock_pipeline_data(email='invalid@yopmail.com'),
{
'SKIP_EMAIL_VALIDATION': False, 'AUTOMATIC_AUTH_FOR_TESTING': False,
'BYPASS_ACTIVATION_EMAIL_FOR_EXTAUTH': False,
},
False # Send activation email when `skip_email_verification` is not set.
)
@@ -551,7 +469,7 @@ class TestCreateAccount(SiteMixin, TestCase):
@ddt.unpack
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_should_skip_activation_email(
self, do_external_auth, skip_email_verification, running_pipeline, feature_overrides, expected,
self, skip_email_verification, running_pipeline, feature_overrides, expected,
):
"""
Test `skip_activation_email` works as expected.
@@ -564,7 +482,6 @@ class TestCreateAccount(SiteMixin, TestCase):
with override_settings(FEATURES=dict(settings.FEATURES, **feature_overrides)):
result = _skip_activation_email(
user=user,
do_external_auth=do_external_auth,
running_pipeline=running_pipeline,
third_party_provider=third_party_provider
)