Add audit log definition, and use for logging of logins in external_auth and student apps. Move test_login to student app. Improve conditional tests for Shibboleth login logic. (Does not include reconfiguring log settings.)
This commit is contained in:
@@ -3,6 +3,7 @@ Tests for Shibboleth Authentication
|
||||
@jbau
|
||||
"""
|
||||
import unittest
|
||||
from mock import patch
|
||||
|
||||
from django.conf import settings
|
||||
from django.http import HttpResponseRedirect
|
||||
@@ -10,7 +11,6 @@ from django.test.client import RequestFactory, Client as DjangoTestClient
|
||||
from django.test.utils import override_settings
|
||||
from django.core.urlresolvers import reverse
|
||||
from django.contrib.auth.models import AnonymousUser, User
|
||||
from django.contrib.sessions.backends.base import SessionBase
|
||||
from django.utils.importlib import import_module
|
||||
|
||||
from xmodule.modulestore.tests.factories import CourseFactory
|
||||
@@ -27,11 +27,11 @@ from student.views import create_account, change_enrollment
|
||||
from student.models import UserProfile, Registration, CourseEnrollment
|
||||
from student.tests.factories import UserFactory
|
||||
|
||||
#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
|
||||
# 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
|
||||
# For the sake of python convention we'll make all of these variable names ALL_CAPS
|
||||
IDP = 'https://idp.stanford.edu/'
|
||||
REMOTE_USER = 'test_user@stanford.edu'
|
||||
MAILS = [None, '', 'test_user@stanford.edu']
|
||||
@@ -93,6 +93,13 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
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"""
|
||||
method_name, args, _kwargs = audit_log_call
|
||||
self.assertEquals(method_name, 'info')
|
||||
self.assertEquals(len(args), 2)
|
||||
self.assertIn(u'logged in via Shibboleth', args[0])
|
||||
self.assertEquals(remote_user, args[1])
|
||||
|
||||
@unittest.skipUnless(settings.MITX_FEATURES.get('AUTH_USE_SHIB'), True)
|
||||
def test_shib_login(self):
|
||||
@@ -140,26 +147,57 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
'REMOTE_USER': remote_user,
|
||||
'mail': remote_user})
|
||||
request.user = AnonymousUser()
|
||||
response = shib_login(request)
|
||||
with patch('external_auth.views.AUDIT_LOG') as mock_audit_log:
|
||||
response = shib_login(request)
|
||||
audit_log_calls = mock_audit_log.method_calls
|
||||
|
||||
if idp == "https://idp.stanford.edu/" and remote_user == 'withmap@stanford.edu':
|
||||
self.assertIsInstance(response, HttpResponseRedirect)
|
||||
self.assertEqual(request.user, user_w_map)
|
||||
self.assertEqual(response['Location'], '/')
|
||||
# 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), 3)
|
||||
self.assertIn(u'Login success', args[0])
|
||||
self.assertEquals(remote_user, args[2])
|
||||
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), 2)
|
||||
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.assertIsInstance(response, HttpResponseRedirect)
|
||||
self.assertEqual(request.user, user_wo_map)
|
||||
self.assertEqual(response['Location'], '/')
|
||||
# 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), 3)
|
||||
self.assertIn(u'Login success', args[0])
|
||||
self.assertEquals(remote_user, args[2])
|
||||
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, "<title>Register for")
|
||||
# no audit logging calls
|
||||
self.assertEquals(len(audit_log_calls), 0)
|
||||
|
||||
@unittest.skipUnless(settings.MITX_FEATURES.get('AUTH_USE_SHIB'), True)
|
||||
def test_registration_form(self):
|
||||
@@ -187,7 +225,7 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
else:
|
||||
self.assertNotContains(response, fullname_input_HTML)
|
||||
|
||||
#clean up b/c we don't want existing ExternalAuthMap for the next run
|
||||
# clean up b/c we don't want existing ExternalAuthMap for the next run
|
||||
client.session['ExternalAuthMap'].delete()
|
||||
|
||||
@unittest.skipUnless(settings.MITX_FEATURES.get('AUTH_USE_SHIB'), True)
|
||||
@@ -200,25 +238,47 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
Uses django test client for its session support
|
||||
"""
|
||||
for identity in gen_all_identities():
|
||||
#First we pop the registration form
|
||||
# First we pop the registration form
|
||||
client = DjangoTestClient()
|
||||
response1 = client.get(path='/shib-login/', data={}, follow=False, **identity)
|
||||
#Then we have the user answer the registration form
|
||||
# Then we have the user answer the registration form
|
||||
postvars = {'email': 'post_email@stanford.edu',
|
||||
'username': 'post_username',
|
||||
'password': 'post_password',
|
||||
'name': 'post_name',
|
||||
'terms_of_service': 'true',
|
||||
'honor_code': 'true'}
|
||||
#use RequestFactory instead of TestClient here because we want access to request.user
|
||||
# use RequestFactory instead of TestClient here because we want access to request.user
|
||||
request2 = self.request_factory.post('/create_account', data=postvars)
|
||||
request2.session = client.session
|
||||
request2.user = AnonymousUser()
|
||||
response2 = create_account(request2)
|
||||
with patch('student.views.AUDIT_LOG') as mock_audit_log:
|
||||
_response2 = create_account(request2)
|
||||
|
||||
user = request2.user
|
||||
mail = identity.get('mail')
|
||||
#check that the created user has the right email, either taken from shib or user input
|
||||
|
||||
# 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), 1)
|
||||
self.assertIn(u'Login success on new account creation', args[0])
|
||||
self.assertIn(u'post_username', args[0])
|
||||
method_name, args, _kwargs = audit_log_calls[1]
|
||||
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[2]
|
||||
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)
|
||||
|
||||
# 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'])), [])
|
||||
@@ -228,7 +288,7 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
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
|
||||
# check that the created user profile has the right name, either taken from shib or user input
|
||||
profile = UserProfile.objects.get(user=user)
|
||||
sn_empty = not identity.get('sn')
|
||||
given_name_empty = not identity.get('givenName')
|
||||
@@ -236,7 +296,7 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
self.assertEqual(profile.name, postvars['name'])
|
||||
else:
|
||||
self.assertEqual(profile.name, request2.session['ExternalAuthMap'].external_name)
|
||||
#clean up for next loop
|
||||
# clean up for next loop
|
||||
request2.session['ExternalAuthMap'].delete()
|
||||
UserProfile.objects.filter(user=user).delete()
|
||||
Registration.objects.filter(user=user).delete()
|
||||
@@ -251,17 +311,17 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
|
||||
# Test for cases where course is found
|
||||
for domain in ["", "shib:https://idp.stanford.edu/"]:
|
||||
#set domains
|
||||
# set domains
|
||||
course.enrollment_domain = domain
|
||||
metadata = own_metadata(course)
|
||||
metadata['enrollment_domain'] = domain
|
||||
self.store.update_metadata(course.location.url(), metadata)
|
||||
|
||||
#setting location to test that GET params get passed through
|
||||
# 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' +
|
||||
_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')
|
||||
|
||||
@@ -292,11 +352,11 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
'&enrollment_action=enroll')
|
||||
|
||||
# Now test for non-existent course
|
||||
#setting location to test that GET params get passed through
|
||||
# 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' +
|
||||
_reg_request = self.request_factory.get('/course_specific_register/DNE/DNE/DNE' +
|
||||
'?course_id=DNE/DNE/DNE/Robot_Super_Course' +
|
||||
'&enrollment_action=enroll')
|
||||
|
||||
@@ -321,7 +381,7 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
the proper external auth
|
||||
"""
|
||||
|
||||
#create 2 course, one with limited enrollment one without
|
||||
# create 2 course, one with limited enrollment one without
|
||||
shib_course = CourseFactory.create(org='Stanford', number='123', display_name='Shib Only')
|
||||
shib_course.enrollment_domain = 'shib:https://idp.stanford.edu/'
|
||||
metadata = own_metadata(shib_course)
|
||||
@@ -360,7 +420,7 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
int_student.email = "teststudent3@gmail.com"
|
||||
int_student.save()
|
||||
|
||||
#Tests the two case for courses, limited and not
|
||||
# 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')
|
||||
@@ -368,11 +428,11 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
'course_id': 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 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.assertEqual(CourseEnrollment.objects.filter(user=student, course_id=course.id).count(), 1)
|
||||
#clean up
|
||||
# Clean up
|
||||
CourseEnrollment.objects.filter(user=student, course_id=course.id).delete()
|
||||
else:
|
||||
self.assertEqual(response.status_code, 400)
|
||||
@@ -383,9 +443,6 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
"""
|
||||
A functionality test that a student with an existing shib login can auto-enroll in a class with GET params
|
||||
"""
|
||||
if not settings.MITX_FEATURES.get('AUTH_USE_SHIB'):
|
||||
return
|
||||
|
||||
student = UserFactory.create()
|
||||
extauth = ExternalAuthMap(external_id='testuser@stanford.edu',
|
||||
external_email='',
|
||||
@@ -403,8 +460,8 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
metadata['enrollment_domain'] = course.enrollment_domain
|
||||
self.store.update_metadata(course.location.url(), metadata)
|
||||
|
||||
#use django test client for sessions and url processing
|
||||
#no enrollment before trying
|
||||
# use django test client for sessions and url processing
|
||||
# no enrollment before trying
|
||||
self.assertEqual(CourseEnrollment.objects.filter(user=student, course_id=course.id).count(), 0)
|
||||
self.client.logout()
|
||||
request_kwargs = {'path': '/shib-login/',
|
||||
@@ -413,8 +470,8 @@ class ShibSPTest(ModuleStoreTestCase):
|
||||
'REMOTE_USER': 'testuser@stanford.edu',
|
||||
'Shib-Identity-Provider': 'https://idp.stanford.edu/'}
|
||||
response = self.client.get(**request_kwargs)
|
||||
#successful login is a redirect to "/"
|
||||
# successful login is a redirect to "/"
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertEqual(response['location'], 'http://testserver/')
|
||||
#now there is enrollment
|
||||
# now there is enrollment
|
||||
self.assertEqual(CourseEnrollment.objects.filter(user=student, course_id=course.id).count(), 1)
|
||||
|
||||
@@ -17,9 +17,9 @@ from django.core.urlresolvers import reverse
|
||||
from django.core.validators import validate_email
|
||||
from django.core.exceptions import ValidationError
|
||||
|
||||
from student.models import UserProfile, TestCenterUser, TestCenterRegistration
|
||||
from student.models import TestCenterUser, TestCenterRegistration
|
||||
|
||||
from django.http import HttpResponse, HttpResponseRedirect, HttpRequest
|
||||
from django.http import HttpResponse, HttpResponseRedirect, HttpRequest, HttpResponseForbidden
|
||||
from django.utils.http import urlquote
|
||||
from django.shortcuts import redirect
|
||||
from django.utils.translation import ugettext as _
|
||||
@@ -50,7 +50,10 @@ from xmodule.modulestore import Location
|
||||
from xmodule.modulestore.exceptions import ItemNotFoundError
|
||||
|
||||
log = logging.getLogger("mitx.external_auth")
|
||||
AUDIT_LOG = logging.getLogger("audit")
|
||||
|
||||
SHIBBOLETH_DOMAIN_PREFIX = 'shib:'
|
||||
OPENID_DOMAIN_PREFIX = 'openid:'
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# OpenID Common
|
||||
@@ -81,7 +84,7 @@ def default_render_failure(request,
|
||||
def generate_password(length=12, chars=string.letters + string.digits):
|
||||
"""Generate internal password for externally authenticated user"""
|
||||
choice = random.SystemRandom().choice
|
||||
return ''.join([choice(chars) for i in range(length)])
|
||||
return ''.join([choice(chars) for _i in range(length)])
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
@@ -105,27 +108,29 @@ def openid_login_complete(request,
|
||||
log.debug('openid success, details=%s', details)
|
||||
|
||||
url = getattr(settings, 'OPENID_SSO_SERVER_URL', None)
|
||||
external_domain = "openid:%s" % url
|
||||
external_domain = "{0}{1}".format(OPENID_DOMAIN_PREFIX, url)
|
||||
fullname = '%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)
|
||||
return _external_login_or_signup(
|
||||
request,
|
||||
external_id,
|
||||
external_domain,
|
||||
details,
|
||||
details.get('email', ''),
|
||||
fullname
|
||||
)
|
||||
|
||||
return render_failure(request, 'Openid failure')
|
||||
|
||||
|
||||
def external_login_or_signup(request,
|
||||
external_id,
|
||||
external_domain,
|
||||
credentials,
|
||||
email,
|
||||
fullname,
|
||||
retfun=None):
|
||||
def _external_login_or_signup(request,
|
||||
external_id,
|
||||
external_domain,
|
||||
credentials,
|
||||
email,
|
||||
fullname,
|
||||
retfun=None):
|
||||
"""Generic external auth login or signup"""
|
||||
|
||||
# see if we have a map from this external_id to an edX username
|
||||
@@ -142,13 +147,13 @@ def external_login_or_signup(request,
|
||||
eamap.external_name = fullname
|
||||
eamap.internal_password = generate_password()
|
||||
log.debug('Created eamap=%s', eamap)
|
||||
|
||||
eamap.save()
|
||||
|
||||
log.info(u"External_Auth login_or_signup for %s : %s : %s : %s", external_domain, external_id, email, fullname)
|
||||
uses_shibboleth = settings.MITX_FEATURES.get('AUTH_USE_SHIB') and external_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX)
|
||||
internal_user = eamap.user
|
||||
if internal_user is None:
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB'):
|
||||
if uses_shibboleth:
|
||||
# if we are using shib, try to link accounts using email
|
||||
try:
|
||||
link_user = User.objects.get(email=eamap.external_email)
|
||||
@@ -169,14 +174,14 @@ def external_login_or_signup(request,
|
||||
return default_render_failure(request, failure_msg)
|
||||
except User.DoesNotExist:
|
||||
log.info('SHIB: No user for %s yet, doing signup', eamap.external_email)
|
||||
return signup(request, eamap)
|
||||
return _signup(request, eamap)
|
||||
else:
|
||||
log.info('No user for %s yet. doing signup', eamap.external_email)
|
||||
return signup(request, eamap)
|
||||
return _signup(request, eamap)
|
||||
|
||||
# We trust shib's authentication, so no need to authenticate using the password again
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB'):
|
||||
uname = internal_user.username
|
||||
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:
|
||||
@@ -184,32 +189,32 @@ def external_login_or_signup(request,
|
||||
else:
|
||||
auth_backend = 'django.contrib.auth.backends.ModelBackend'
|
||||
user.backend = auth_backend
|
||||
log.info('SHIB: Logging in linked user %s', user.email)
|
||||
AUDIT_LOG.info('Linked user "%s" logged in via Shibboleth', user.email)
|
||||
else:
|
||||
uname = internal_user.username
|
||||
user = authenticate(username=uname, password=eamap.internal_password)
|
||||
if user is None:
|
||||
log.warning("External Auth Login failed for %s / %s",
|
||||
uname, eamap.internal_password)
|
||||
return signup(request, eamap)
|
||||
# we want to log the failure, but don't want to log the password attempted:
|
||||
AUDIT_LOG.warning('External Auth Login failed for "%s"', uname)
|
||||
return _signup(request, eamap)
|
||||
|
||||
if not user.is_active:
|
||||
log.warning("User %s is not active", uname)
|
||||
AUDIT_LOG.warning('User "%s" is not active after external login', 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)
|
||||
|
||||
# Now to try enrollment
|
||||
# Need to special case Shibboleth here because it logs in via a GET.
|
||||
# testing request.method for extra paranoia
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB') and 'shib:' in external_domain and request.method == 'GET':
|
||||
enroll_request = make_shib_enrollment_request(request)
|
||||
if uses_shibboleth and request.method == 'GET':
|
||||
enroll_request = _make_shib_enrollment_request(request)
|
||||
student_views.try_change_enrollment(enroll_request)
|
||||
else:
|
||||
student_views.try_change_enrollment(request)
|
||||
log.info("Login success - %s (%s)", user.username, user.email)
|
||||
AUDIT_LOG.info("Login success - %s (%s)", user.username, user.email)
|
||||
if retfun is None:
|
||||
return redirect('/')
|
||||
return retfun()
|
||||
@@ -217,20 +222,16 @@ def external_login_or_signup(request,
|
||||
|
||||
@ensure_csrf_cookie
|
||||
@cache_if_anonymous
|
||||
def signup(request, eamap=None):
|
||||
def _signup(request, eamap):
|
||||
"""
|
||||
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 ExteralAuthMap object, specifying the external user
|
||||
eamap is an ExternalAuthMap object, specifying the external user
|
||||
for which to complete the signup.
|
||||
"""
|
||||
|
||||
if eamap is None:
|
||||
pass
|
||||
|
||||
# save this for use by student.views.create_account
|
||||
request.session['ExternalAuthMap'] = eamap
|
||||
|
||||
@@ -248,8 +249,9 @@ def signup(request, eamap=None):
|
||||
|
||||
# Some openEdX instances can't have terms of service for shib users, like
|
||||
# according to Stanford's Office of General Counsel
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB') and settings.MITX_FEATURES.get('SHIB_DISABLE_TOS') and \
|
||||
('shib' in eamap.external_domain):
|
||||
uses_shibboleth = (settings.MITX_FEATURES.get('AUTH_USE_SHIB') and
|
||||
eamap.external_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX))
|
||||
if uses_shibboleth and settings.MITX_FEATURES.get('SHIB_DISABLE_TOS'):
|
||||
context['ask_for_tos'] = False
|
||||
|
||||
# detect if full name is blank and ask for it from user
|
||||
@@ -272,19 +274,19 @@ def signup(request, eamap=None):
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
def ssl_dn_extract_info(dn):
|
||||
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.
|
||||
"""
|
||||
ss = re.search('/emailAddress=(.*)@([^/]+)', dn)
|
||||
ss = re.search('/emailAddress=(.*)@([^/]+)', dn_string)
|
||||
if ss:
|
||||
user = ss.group(1)
|
||||
email = "%s@%s" % (user, ss.group(2))
|
||||
else:
|
||||
return None
|
||||
ss = re.search('/CN=([^/]+)/', dn)
|
||||
ss = re.search('/CN=([^/]+)/', dn_string)
|
||||
if ss:
|
||||
fullname = ss.group(1)
|
||||
else:
|
||||
@@ -292,7 +294,7 @@ def ssl_dn_extract_info(dn):
|
||||
return (user, email, fullname)
|
||||
|
||||
|
||||
def ssl_get_cert_from_request(request):
|
||||
def _ssl_get_cert_from_request(request):
|
||||
"""
|
||||
Extract user information from certificate, if it exists, returning (user, email, fullname).
|
||||
Else return None.
|
||||
@@ -311,9 +313,6 @@ def ssl_get_cert_from_request(request):
|
||||
|
||||
return cert
|
||||
|
||||
(user, email, fullname) = ssl_dn_extract_info(cert)
|
||||
return (user, email, fullname)
|
||||
|
||||
|
||||
def ssl_login_shortcut(fn):
|
||||
"""
|
||||
@@ -324,24 +323,26 @@ def ssl_login_shortcut(fn):
|
||||
if not settings.MITX_FEATURES['AUTH_USE_MIT_CERTIFICATES']:
|
||||
return fn(*args, **kwargs)
|
||||
request = args[0]
|
||||
cert = ssl_get_cert_from_request(request)
|
||||
cert = _ssl_get_cert_from_request(request)
|
||||
if not cert: # no certificate information - show normal login window
|
||||
return fn(*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)
|
||||
(_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
|
||||
)
|
||||
return wrapped
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def ssl_login(request):
|
||||
"""
|
||||
This is called by student.views.index when
|
||||
This is called by branding.views.index when
|
||||
MITX_FEATURES['AUTH_USE_MIT_CERTIFICATES'] = True
|
||||
|
||||
Used for MIT user authentication. This presumes the web server
|
||||
@@ -355,22 +356,28 @@ def ssl_login(request):
|
||||
|
||||
Else continues on with student.views.index, and no authentication.
|
||||
"""
|
||||
cert = ssl_get_cert_from_request(request)
|
||||
# Just to make sure we're calling this only at MIT:
|
||||
if not settings.MITX_FEATURES['AUTH_USE_MIT_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)
|
||||
(_user, email, fullname) = _ssl_dn_extract_info(cert)
|
||||
|
||||
retfun = functools.partial(student_views.index, request)
|
||||
return external_login_or_signup(request,
|
||||
external_id=email,
|
||||
external_domain="ssl:MIT",
|
||||
credentials=cert,
|
||||
email=email,
|
||||
fullname=fullname,
|
||||
retfun=retfun)
|
||||
return _external_login_or_signup(
|
||||
request,
|
||||
external_id=email,
|
||||
external_domain="ssl:MIT",
|
||||
credentials=cert,
|
||||
email=email,
|
||||
fullname=fullname,
|
||||
retfun=retfun
|
||||
)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@@ -396,28 +403,30 @@ def shib_login(request):
|
||||
log.error("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
|
||||
# If we get here, the user has authenticated properly
|
||||
shib = {attr: request.META.get(attr, '')
|
||||
for attr in ['REMOTE_USER', 'givenName', 'sn', 'mail', 'Shib-Identity-Provider']}
|
||||
|
||||
#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
|
||||
# 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().decode('utf-8')
|
||||
shib['givenName'] = shib['givenName'].split(";")[0].strip().capitalize().decode('utf-8')
|
||||
|
||||
# TODO: should we be logging creds here, at info level?
|
||||
log.info("SHIB creds returned: %r", shib)
|
||||
|
||||
return external_login_or_signup(request,
|
||||
external_id=shib['REMOTE_USER'],
|
||||
external_domain="shib:" + shib['Shib-Identity-Provider'],
|
||||
credentials=shib,
|
||||
email=shib['mail'],
|
||||
fullname=u'%s %s' % (shib['givenName'], shib['sn']),
|
||||
)
|
||||
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=u'%s %s' % (shib['givenName'], shib['sn']),
|
||||
)
|
||||
|
||||
|
||||
def make_shib_enrollment_request(request):
|
||||
def _make_shib_enrollment_request(request):
|
||||
"""
|
||||
Need this hack function because shibboleth logins don't happen over POST
|
||||
but change_enrollment expects its request to be a POST, with
|
||||
@@ -452,14 +461,14 @@ def course_specific_login(request, course_id):
|
||||
try:
|
||||
course = course_from_id(course_id)
|
||||
except ItemNotFoundError:
|
||||
#couldn't find the course, will just return vanilla signin page
|
||||
# couldn't find the course, will just return vanilla signin page
|
||||
return redirect_with_querystring('signin_user', query_string)
|
||||
|
||||
#now the dispatching conditionals. Only shib for now
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB') and 'shib:' in course.enrollment_domain:
|
||||
# now the dispatching conditionals. Only shib for now
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB') and course.enrollment_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX):
|
||||
return redirect_with_querystring('shib-login', query_string)
|
||||
|
||||
#Default fallthrough to normal signin page
|
||||
# Default fallthrough to normal signin page
|
||||
return redirect_with_querystring('signin_user', query_string)
|
||||
|
||||
|
||||
@@ -473,15 +482,15 @@ def course_specific_register(request, course_id):
|
||||
try:
|
||||
course = course_from_id(course_id)
|
||||
except ItemNotFoundError:
|
||||
#couldn't find the course, will just return vanilla registration page
|
||||
# couldn't find the course, will just return vanilla registration page
|
||||
return redirect_with_querystring('register_user', query_string)
|
||||
|
||||
#now the dispatching conditionals. Only shib for now
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB') and 'shib:' in course.enrollment_domain:
|
||||
#shib-login takes care of both registration and login flows
|
||||
# now the dispatching conditionals. Only shib for now
|
||||
if settings.MITX_FEATURES.get('AUTH_USE_SHIB') and course.enrollment_domain.startswith(SHIBBOLETH_DOMAIN_PREFIX):
|
||||
# shib-login takes care of both registration and login flows
|
||||
return redirect_with_querystring('shib-login', query_string)
|
||||
|
||||
#Default fallthrough to normal registration page
|
||||
# Default fallthrough to normal registration page
|
||||
return redirect_with_querystring('register_user', query_string)
|
||||
|
||||
|
||||
@@ -702,7 +711,7 @@ def provider_login(request):
|
||||
except User.DoesNotExist:
|
||||
request.session['openid_error'] = True
|
||||
msg = "OpenID login failed - Unknown user email: %s"
|
||||
log.warning(msg, email)
|
||||
AUDIT_LOG.warning(msg, email)
|
||||
return HttpResponseRedirect(openid_request_url)
|
||||
|
||||
# attempt to authenticate user (but not actually log them in...)
|
||||
@@ -713,7 +722,7 @@ def provider_login(request):
|
||||
if user is None:
|
||||
request.session['openid_error'] = True
|
||||
msg = "OpenID login failed - password for %s is invalid"
|
||||
log.warning(msg, email)
|
||||
AUDIT_LOG.warning(msg, email)
|
||||
return HttpResponseRedirect(openid_request_url)
|
||||
|
||||
# authentication succeeded, so fetch user information
|
||||
@@ -723,8 +732,8 @@ def provider_login(request):
|
||||
if 'openid_error' in request.session:
|
||||
del request.session['openid_error']
|
||||
|
||||
log.info("OpenID login success - %s (%s)",
|
||||
user.username, user.email)
|
||||
AUDIT_LOG.info("OpenID login success - %s (%s)",
|
||||
user.username, user.email)
|
||||
|
||||
# redirect user to return_to location
|
||||
url = endpoint + urlquote(user.username)
|
||||
@@ -755,7 +764,7 @@ def provider_login(request):
|
||||
# the account is not active, so redirect back to the login page:
|
||||
request.session['openid_error'] = True
|
||||
msg = "Login failed - Account not active for user %s"
|
||||
log.warning(msg, username)
|
||||
AUDIT_LOG.warning(msg, username)
|
||||
return HttpResponseRedirect(openid_request_url)
|
||||
|
||||
# determine consumer domain if applicable
|
||||
@@ -856,9 +865,11 @@ def test_center_login(request):
|
||||
try:
|
||||
testcenteruser = TestCenterUser.objects.get(client_candidate_id=client_candidate_id)
|
||||
except TestCenterUser.DoesNotExist:
|
||||
log.error("not able to find demographics for cand ID {}".format(client_candidate_id))
|
||||
AUDIT_LOG.error("not able to find demographics for cand ID {}".format(client_candidate_id))
|
||||
return HttpResponseRedirect(makeErrorURL(error_url, "invalidClientCandidateID"))
|
||||
|
||||
AUDIT_LOG.info("Attempting to log in test-center user '{}' for test of cand {}".format(testcenteruser.user.username, client_candidate_id))
|
||||
|
||||
# find testcenter_registration that matches the provided exam code:
|
||||
# Note that we could rely in future on either the registrationId or the exam code,
|
||||
# or possibly both. But for now we know what to do with an ExamSeriesCode,
|
||||
@@ -867,13 +878,13 @@ def test_center_login(request):
|
||||
# we are not allowed to make up a new error code, according to Pearson,
|
||||
# so instead of "missingExamSeriesCode", we use a valid one that is
|
||||
# inaccurate but at least distinct. (Sigh.)
|
||||
log.error("missing exam series code for cand ID {}".format(client_candidate_id))
|
||||
AUDIT_LOG.error("missing exam series code for cand ID {}".format(client_candidate_id))
|
||||
return HttpResponseRedirect(makeErrorURL(error_url, "missingPartnerID"))
|
||||
exam_series_code = request.POST.get('vueExamSeriesCode')
|
||||
|
||||
registrations = TestCenterRegistration.objects.filter(testcenter_user=testcenteruser, exam_series_code=exam_series_code)
|
||||
if not registrations:
|
||||
log.error("not able to find exam registration for exam {} and cand ID {}".format(exam_series_code, client_candidate_id))
|
||||
AUDIT_LOG.error("not able to find exam registration for exam {} and cand ID {}".format(exam_series_code, client_candidate_id))
|
||||
return HttpResponseRedirect(makeErrorURL(error_url, "noTestsAssigned"))
|
||||
|
||||
# TODO: figure out what to do if there are more than one registrations....
|
||||
@@ -883,14 +894,14 @@ def test_center_login(request):
|
||||
course_id = registration.course_id
|
||||
course = course_from_id(course_id) # assume it will be found....
|
||||
if not course:
|
||||
log.error("not able to find course from ID {} for cand ID {}".format(course_id, client_candidate_id))
|
||||
AUDIT_LOG.error("not able to find course from ID {} for cand ID {}".format(course_id, client_candidate_id))
|
||||
return HttpResponseRedirect(makeErrorURL(error_url, "incorrectCandidateTests"))
|
||||
exam = course.get_test_center_exam(exam_series_code)
|
||||
if not exam:
|
||||
log.error("not able to find exam {} for course ID {} and cand ID {}".format(exam_series_code, course_id, client_candidate_id))
|
||||
AUDIT_LOG.error("not able to find exam {} for course ID {} and cand ID {}".format(exam_series_code, course_id, client_candidate_id))
|
||||
return HttpResponseRedirect(makeErrorURL(error_url, "incorrectCandidateTests"))
|
||||
location = exam.exam_url
|
||||
log.info("proceeding with test of cand {} on exam {} for course {}: URL = {}".format(client_candidate_id, exam_series_code, course_id, location))
|
||||
log.info("Proceeding with test of cand {} on exam {} for course {}: URL = {}".format(client_candidate_id, exam_series_code, course_id, location))
|
||||
|
||||
# check if the test has already been taken
|
||||
timelimit_descriptor = modulestore().get_instance(course_id, Location(location))
|
||||
@@ -907,7 +918,7 @@ def test_center_login(request):
|
||||
return HttpResponseRedirect(makeErrorURL(error_url, "missingClientProgram"))
|
||||
|
||||
if timelimit_module and timelimit_module.has_ended:
|
||||
log.warning("cand {} on exam {} for course {}: test already over at {}".format(client_candidate_id, exam_series_code, course_id, timelimit_module.ending_at))
|
||||
AUDIT_LOG.warning("cand {} on exam {} for course {}: test already over at {}".format(client_candidate_id, exam_series_code, course_id, timelimit_module.ending_at))
|
||||
return HttpResponseRedirect(makeErrorURL(error_url, "allTestsTaken"))
|
||||
|
||||
# check if we need to provide an accommodation:
|
||||
@@ -922,7 +933,7 @@ def test_center_login(request):
|
||||
|
||||
if time_accommodation_code:
|
||||
timelimit_module.accommodation_code = time_accommodation_code
|
||||
log.info("cand {} on exam {} for course {}: receiving accommodation {}".format(client_candidate_id, exam_series_code, course_id, time_accommodation_code))
|
||||
AUDIT_LOG.info("cand {} on exam {} for course {}: receiving accommodation {}".format(client_candidate_id, exam_series_code, course_id, time_accommodation_code))
|
||||
|
||||
# UGLY HACK!!!
|
||||
# Login assumes that authentication has occurred, and that there is a
|
||||
@@ -936,6 +947,7 @@ def test_center_login(request):
|
||||
# testcenteruser.user.backend = "%s.%s" % (backend.__module__, backend.__class__.__name__)
|
||||
testcenteruser.user.backend = "%s.%s" % ("TestcenterAuthenticationModule", "TestcenterAuthenticationClass")
|
||||
login(request, testcenteruser.user)
|
||||
AUDIT_LOG.info("Logged in user '{}' for test of cand {} on exam {} for course {}: URL = {}".format(testcenteruser.user.username, client_candidate_id, exam_series_code, course_id, location))
|
||||
|
||||
# And start the test:
|
||||
return jump_to(request, course_id, location)
|
||||
|
||||
Reference in New Issue
Block a user