Skip to content

D
django-openid-auth
Commit f767f79e by Michael Hall
Options

Merge from trunk

parents e09735ff fe78850c
Showing with 171 additions and 69 deletions
django_openid_auth/auth.py
......@@ -42,8 +42,9 @@ from django_openid_auth.exceptions import (
DuplicateUsernameViolation,
MissingUsernameViolation,
MissingPhysicalMultiFactor,
RequiredAttributeNotReturned,
)
class OpenIDBackend:
"""A django.contrib.auth backend that authenticates the user based on
an OpenID response."""
......@@ -134,8 +135,10 @@ class OpenIDBackend:
if fullname and not (first_name or last_name):
# Django wants to store first and last names separately,
# so we do our best to split the full name.
if ' ' in fullname:
first_name, last_name = fullname.rsplit(None, 1)
fullname = fullname.strip()
split_names = fullname.rsplit(None, 1)
if len(split_names) == 2:
first_name, last_name = split_names
else:
first_name = u''
last_name = fullname
......@@ -159,7 +162,7 @@ class OpenIDBackend:
except User.DoesNotExist:
# No conflict, we can use this nickname
return nickname
# Check if we already have nickname+i for this identity_url
try:
user_openid = UserOpenID.objects.get(
......@@ -180,7 +183,7 @@ class OpenIDBackend:
except UserOpenID.DoesNotExist:
# No user associated with this identity_url
pass
if getattr(settings, 'OPENID_STRICT_USERNAMES', False):
if User.objects.filter(username__exact=nickname).count() > 0:
......@@ -199,9 +202,19 @@ class OpenIDBackend:
break
i += 1
return username
def create_user_from_openid(self, openid_response):
details = self._extract_user_details(openid_response)
required_attrs = getattr(settings, 'OPENID_SREG_REQUIRED_FIELDS', [])
if getattr(settings, 'OPENID_STRICT_USERNAMES', False):
required_attrs.append('nickname')
for required_attr in required_attrs:
if required_attr not in details or not details[required_attr]:
raise RequiredAttributeNotReturned(
"An attribute required for logging in was not "
"returned ({0}).".format(required_attr))
nickname = details['nickname'] or 'openiduser'
email = details['email'] or ''
......@@ -236,10 +249,10 @@ class OpenIDBackend:
def update_user_details(self, user, details, openid_response):
updated = False
if details['first_name']:
user.first_name = details['first_name']
user.first_name = details['first_name'][:30]
updated = True
if details['last_name']:
user.last_name = details['last_name']
user.last_name = details['last_name'][:30]
updated = True
if details['email']:
user.email = details['email']
......
django_openid_auth/exceptions.py
......@@ -31,6 +31,9 @@
class DjangoOpenIDException(Exception):
pass
class RequiredAttributeNotReturned(DjangoOpenIDException):
pass
class IdentityAlreadyClaimed(DjangoOpenIDException):
def __init__(self, message=None):
......
django_openid_auth/tests/test_auth.py
......@@ -28,6 +28,7 @@
import unittest
from django.contrib.auth.models import User
from django.test import TestCase
from django_openid_auth.auth import OpenIDBackend
......@@ -60,74 +61,95 @@ class OpenIDBackendTests(TestCase):
"last_name": "User",
"email": "foo@example.com"})
def test_extract_user_details_ax(self):
def make_response_ax(self, schema="http://axschema.org/",
fullname="Some User", nickname="someuser", email="foo@example.com",
first=None, last=None):
endpoint = OpenIDServiceEndpoint()
message = Message(OPENID2_NS)
attributes = [
("nickname", "http://axschema.org/namePerson/friendly", "someuser"),
("fullname", "http://axschema.org/namePerson", "Some User"),
("email", "http://axschema.org/contact/email", "foo@example.com"),
("nickname", schema + "namePerson/friendly", nickname),
("fullname", schema + "namePerson", fullname),
("email", schema + "contact/email", email),
]
if first:
attributes.append(
("first", "http://axschema.org/namePerson/first", first))
if last:
attributes.append(
("last", "http://axschema.org/namePerson/last", last))
message.setArg(AX_NS, "mode", "fetch_response")
for (alias, uri, value) in attributes:
message.setArg(AX_NS, "type.%s" % alias, uri)
message.setArg(AX_NS, "value.%s" % alias, value)
response = SuccessResponse(
return SuccessResponse(
endpoint, message, signed_fields=message.toPostArgs().keys())
def test_extract_user_details_ax(self):
response = self.make_response_ax(fullname="Some User",
nickname="someuser", email="foo@example.com")
data = self.backend._extract_user_details(response)
self.assertEqual(data, {"nickname": "someuser",
"first_name": "Some",
"last_name": "User",
"email": "foo@example.com"})
def test_extract_user_details_ax_split_name(self):
endpoint = OpenIDServiceEndpoint()
message = Message(OPENID2_NS)
attributes = [
("nickname", "http://axschema.org/namePerson/friendly", "someuser"),
# Include this key too to show that the split data takes
# precedence.
("fullname", "http://axschema.org/namePerson", "Bad Data"),
("first", "http://axschema.org/namePerson/first", "Some"),
("last", "http://axschema.org/namePerson/last", "User"),
("email", "http://axschema.org/contact/email", "foo@example.com"),
]
message.setArg(AX_NS, "mode", "fetch_response")
for (alias, uri, value) in attributes:
message.setArg(AX_NS, "type.%s" % alias, uri)
message.setArg(AX_NS, "value.%s" % alias, value)
response = SuccessResponse(
endpoint, message, signed_fields=message.toPostArgs().keys())
# Include fullname too to show that the split data takes
# precedence.
response = self.make_response_ax(
fullname="Bad Data", first="Some", last="User")
data = self.backend._extract_user_details(response)
self.assertEqual(data, {"nickname": "someuser",
"first_name": "Some",
"last_name": "User",
"email": "foo@example.com"})
def test_extract_user_details_ax_broken_myopenid(self):
endpoint = OpenIDServiceEndpoint()
message = Message(OPENID2_NS)
attributes = [
("nickname", "http://schema.openid.net/namePerson/friendly",
"someuser"),
("fullname", "http://schema.openid.net/namePerson", "Some User"),
("email", "http://schema.openid.net/contact/email",
"foo@example.com"),
]
message.setArg(AX_NS, "mode", "fetch_response")
for (alias, uri, value) in attributes:
message.setArg(AX_NS, "type.%s" % alias, uri)
message.setArg(AX_NS, "value.%s" % alias, value)
response = SuccessResponse(
endpoint, message, signed_fields=message.toPostArgs().keys())
response = self.make_response_ax(
schema="http://schema.openid.net/", fullname="Some User",
nickname="someuser", email="foo@example.com")
data = self.backend._extract_user_details(response)
self.assertEqual(data, {"nickname": "someuser",
"first_name": "Some",
"last_name": "User",
"email": "foo@example.com"})
def test_update_user_details_long_names(self):
response = self.make_response_ax()
user = User.objects.create_user('someuser', 'someuser@example.com',
password=None)
data = dict(first_name=u"Some56789012345678901234567890123",
last_name=u"User56789012345678901234567890123",
email=u"someotheruser@example.com")
self.backend.update_user_details(user, data, response)
self.assertEqual("Some56789012345678901234567890", user.first_name)
self.assertEqual("User56789012345678901234567890", user.last_name)
def test_extract_user_details_name_with_trailing_space(self):
response = self.make_response_ax(fullname="SomeUser ")
data = self.backend._extract_user_details(response)
self.assertEqual("", data['first_name'])
self.assertEqual("SomeUser", data['last_name'])
def test_extract_user_details_name_with_thin_space(self):
response = self.make_response_ax(fullname=u"Some\u2009User")
data = self.backend._extract_user_details(response)
self.assertEqual("Some", data['first_name'])
self.assertEqual("User", data['last_name'])
def suite():
return unittest.TestLoader().loadTestsFromName(__name__)
django_openid_auth/tests/test_views.py
......@@ -57,6 +57,7 @@ from django_openid_auth.exceptions import (
MissingUsernameViolation,
DuplicateUsernameViolation,
MissingPhysicalMultiFactor,
RequiredAttributeNotReturned,
)
ET = importElementTree()
......@@ -185,6 +186,8 @@ class RelyingPartyTests(TestCase):
self.old_login_render_failure = getattr(settings, 'OPENID_RENDER_FAILURE', None)
self.old_consumer_complete = Consumer.complete
self.old_required_fields = getattr(
settings, 'OPENID_SREG_REQUIRED_FIELDS', [])
settings.OPENID_CREATE_USERS = False
settings.OPENID_STRICT_USERNAMES = False
......@@ -194,6 +197,7 @@ class RelyingPartyTests(TestCase):
settings.OPENID_USE_AS_ADMIN_LOGIN = False
settings.OPENID_FOLLOW_RENAMES = False
settings.OPENID_PHYSICAL_MULTIFACTOR_REQUIRED = False
settings.OPENID_SREG_REQUIRED_FIELDS = []
def tearDown(self):
settings.LOGIN_REDIRECT_URL = self.old_login_redirect_url
......@@ -207,6 +211,7 @@ class RelyingPartyTests(TestCase):
settings.OPENID_PHYSICAL_MULTIFACTOR_REQUIRED = self.old_physical_multifactor
settings.OPENID_RENDER_FAILURE = self.old_login_render_failure
Consumer.complete = self.old_consumer_complete
settings.OPENID_SREG_REQUIRED_FIELDS = self.old_required_fields
setDefaultFetcher(None)
super(RelyingPartyTests, self).tearDown()
......@@ -488,7 +493,7 @@ class RelyingPartyTests(TestCase):
response = self.complete(openid_response)
self.assertEquals(403, response.status_code)
self.assertContains(response, '<h1>OpenID failed</h1>', status_code=403)
self.assertContains(response, '<p>Login Failed</p>', status_code=403)
self.assertContains(response, '<p>Login requires physical multi-factor authentication.</p>', status_code=403)
def test_login_physical_multifactor_not_provided_override(self):
settings.OPENID_PHYSICAL_MULTIFACTOR_REQUIRED = True
......@@ -499,7 +504,6 @@ class RelyingPartyTests(TestCase):
def mock_login_failure_handler(request, message, status=403,
template_name=None,
exception=None):
self.assertEquals(message, 'Login Failed')
self.assertTrue(isinstance(exception, MissingPhysicalMultiFactor))
return HttpResponse('Test Failure Override', status=200)
settings.OPENID_RENDER_FAILURE = mock_login_failure_handler
......@@ -564,7 +568,7 @@ class RelyingPartyTests(TestCase):
self.assertEquals(user.first_name, 'Openid')
self.assertEquals(user.last_name, 'User')
self.assertEquals(user.email, 'foo@example.com')
def test_login_follow_rename(self):
settings.OPENID_FOLLOW_RENAMES = True
settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
......@@ -582,7 +586,7 @@ class RelyingPartyTests(TestCase):
self._do_user_login(openid_req, openid_resp)
response = self.client.get('/getuser/')
# If OPENID_FOLLOW_RENAMES, they are logged in as
# If OPENID_FOLLOW_RENAMES, they are logged in as
# someuser (the passed in nickname has changed the username)
self.assertEquals(response.content, 'someuser')
......@@ -591,7 +595,7 @@ class RelyingPartyTests(TestCase):
self.assertEquals(user.first_name, 'Some')
self.assertEquals(user.last_name, 'User')
self.assertEquals(user.email, 'foo@example.com')
def test_login_follow_rename_conflict(self):
settings.OPENID_FOLLOW_RENAMES = True
settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
......@@ -628,7 +632,7 @@ class RelyingPartyTests(TestCase):
self.assertEquals(user.first_name, 'Rename')
self.assertEquals(user.last_name, 'User')
self.assertEquals(user.email, 'rename@example.com')
def test_login_follow_rename_false_onlyonce(self):
settings.OPENID_FOLLOW_RENAMES = True
settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
......@@ -658,8 +662,8 @@ class RelyingPartyTests(TestCase):
# If OPENID_FOLLOW_RENAMES, attempt to change username to 'testuser'
# but since that username is already taken by someone else, we go through
# the process of adding +i to it. Even though it looks like the username
# follows the nickname+i scheme, it has non-numbers in the suffix, so
# it's not an auto-generated one. The regular process of renaming to
# follows the nickname+i scheme, it has non-numbers in the suffix, so
# it's not an auto-generated one. The regular process of renaming to
# 'testuser' has a conflict, so we get +2 at the end.
self.assertEquals(response.content, 'testuser2')
......@@ -668,7 +672,7 @@ class RelyingPartyTests(TestCase):
self.assertEquals(user.first_name, 'Rename')
self.assertEquals(user.last_name, 'User')
self.assertEquals(user.email, 'rename@example.com')
def test_login_follow_rename_conflict_onlyonce(self):
settings.OPENID_FOLLOW_RENAMES = True
settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
......@@ -706,7 +710,7 @@ class RelyingPartyTests(TestCase):
self.assertEquals(user.first_name, 'Rename')
self.assertEquals(user.last_name, 'User')
self.assertEquals(user.email, 'rename@example.com')
def test_login_follow_rename_false_conflict(self):
settings.OPENID_FOLLOW_RENAMES = True
settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
......@@ -736,10 +740,11 @@ class RelyingPartyTests(TestCase):
self.assertEquals(user.first_name, 'Same')
self.assertEquals(user.last_name, 'User')
self.assertEquals(user.email, 'same@example.com')
def test_strict_username_no_nickname(self):
settings.OPENID_CREATE_USERS = True
settings.OPENID_STRICT_USERNAMES = True
settings.OPENID_SREG_REQUIRED_FIELDS = []
# Posting in an identity URL begins the authentication request:
response = self.client.post('/openid/login/',
......@@ -758,22 +763,23 @@ class RelyingPartyTests(TestCase):
'email': 'foo@example.com'})
openid_response.addExtension(sreg_response)
response = self.complete(openid_response)
# Status code should be 403: Forbidden
self.assertEquals(403, response.status_code)
self.assertContains(response, '<h1>OpenID failed</h1>', status_code=403)
self.assertContains(response, '<p>Login Failed</p>', status_code=403)
self.assertContains(response, "An attribute required for logging in was not returned "
"(nickname)", status_code=403)
def test_strict_username_no_nickname_override(self):
settings.OPENID_CREATE_USERS = True
settings.OPENID_STRICT_USERNAMES = True
settings.OPENID_SREG_REQUIRED_FIELDS = []
# Override the login_failure handler
def mock_login_failure_handler(request, message, status=403,
template_name=None,
exception=None):
self.assertEquals(message, 'Login Failed')
self.assertTrue(isinstance(exception, MissingUsernameViolation))
self.assertTrue(isinstance(exception, (RequiredAttributeNotReturned, MissingUsernameViolation)))
return HttpResponse('Test Failure Override', status=200)
settings.OPENID_RENDER_FAILURE = mock_login_failure_handler
......@@ -826,11 +832,11 @@ class RelyingPartyTests(TestCase):
'email': 'foo@example.com'})
openid_response.addExtension(sreg_response)
response = self.complete(openid_response)
# Status code should be 403: Forbidden
self.assertEquals(403, response.status_code)
self.assertContains(response, '<h1>OpenID failed</h1>', status_code=403)
self.assertContains(response, '<p>Login Failed</p>', status_code=403)
self.assertContains(response, '<p>Duplicate username: someuser</p>', status_code=403)
def test_strict_username_duplicate_user_override(self):
settings.OPENID_CREATE_USERS = True
......@@ -840,7 +846,6 @@ class RelyingPartyTests(TestCase):
def mock_login_failure_handler(request, message, status=403,
template_name=None,
exception=None):
self.assertEquals(message, 'Login Failed')
self.assertTrue(isinstance(exception, DuplicateUsernameViolation))
return HttpResponse('Test Failure Override', status=200)
settings.OPENID_RENDER_FAILURE = mock_login_failure_handler
......@@ -874,6 +879,35 @@ class RelyingPartyTests(TestCase):
self.assertEquals(200, response.status_code)
self.assertContains(response, 'Test Failure Override')
def test_login_requires_sreg_required_fields(self):
# If any required attributes are not included in the response,
# we fail with a forbidden.
settings.OPENID_CREATE_USERS = True
settings.OPENID_SREG_REQUIRED_FIELDS = ('email', 'language')
# Posting in an identity URL begins the authentication request:
response = self.client.post('/openid/login/',
{'openid_identifier': 'http://example.com/identity',
'next': '/getuser/'})
self.assertContains(response, 'OpenID transaction in progress')
# Complete the request, passing back some simple registration
# data. The user is redirected to the next URL.
openid_request = self.provider.parseFormPost(response.content)
sreg_request = sreg.SRegRequest.fromOpenIDRequest(openid_request)
openid_response = openid_request.answer(True)
sreg_response = sreg.SRegResponse.extractResponse(
sreg_request, {'nickname': 'foo',
'fullname': 'Some User',
'email': 'foo@example.com'})
openid_response.addExtension(sreg_response)
response = self.complete(openid_response)
# Status code should be 403: Forbidden as we didn't include
# a required field - language.
self.assertContains(response,
"An attribute required for logging in was not returned "
"(language)", status_code=403)
def test_login_update_details(self):
settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
user = User.objects.create_user('testuser', 'someone@example.com')
......@@ -918,6 +952,27 @@ class RelyingPartyTests(TestCase):
for field in ('email', 'fullname', 'nickname', 'language'):
self.assertTrue(field in sreg_request)
def test_login_uses_sreg_required_fields(self):
# The configurable sreg attributes are used in the request.
settings.OPENID_SREG_REQUIRED_FIELDS = ('email', 'language')
user = User.objects.create_user('testuser', 'someone@example.com')
useropenid = UserOpenID(
user=user,
claimed_id='http://example.com/identity',
display_id='http://example.com/identity')
useropenid.save()
# Posting in an identity URL begins the authentication request:
response = self.client.post('/openid/login/',
{'openid_identifier': 'http://example.com/identity',
'next': '/getuser/'})
openid_request = self.provider.parseFormPost(response.content)
sreg_request = sreg.SRegRequest.fromOpenIDRequest(openid_request)
self.assertEqual(['email', 'language'], sreg_request.required)
self.assertEqual(['fullname', 'nickname'], sreg_request.optional)
def test_login_attribute_exchange(self):
settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
user = User.objects.create_user('testuser', 'someone@example.com')
......
django_openid_auth/views.py
......@@ -55,7 +55,10 @@ from django_openid_auth.forms import OpenIDLoginForm
from django_openid_auth.models import UserOpenID
from django_openid_auth.signals import openid_login_complete
from django_openid_auth.store import DjangoOpenIDStore
from django_openid_auth.exceptions import DjangoOpenIDException
from django_openid_auth.exceptions import (
RequiredAttributeNotReturned,
DjangoOpenIDException,
)
next_url_re = re.compile('^/[-\w/]+$')
......@@ -199,11 +202,18 @@ def login_begin(request, template_name='openid/login.html',
fetch_request.add(ax.AttrInfo(attr, alias=alias, required=True))
openid_request.addExtension(fetch_request)
else:
sreg_required_fields = []
sreg_required_fields.extend(
getattr(settings, 'OPENID_SREG_REQUIRED_FIELDS', []))
sreg_optional_fields = ['email', 'fullname', 'nickname']
extra_fields = getattr(settings, 'OPENID_SREG_EXTRA_FIELDS', [])
sreg_optional_fields.extend(extra_fields)
sreg_optional_fields.extend(
getattr(settings, 'OPENID_SREG_EXTRA_FIELDS', []))
sreg_optional_fields = [
field for field in sreg_optional_fields if (
not field in sreg_required_fields)]
openid_request.addExtension(
sreg.SRegRequest(optional=sreg_optional_fields))
sreg.SRegRequest(optional=sreg_optional_fields,
required=sreg_required_fields))
if getattr(settings, 'OPENID_PHYSICAL_MULTIFACTOR_REQUIRED', False):
preferred_auth = [
......@@ -212,7 +222,6 @@ def login_begin(request, template_name='openid/login.html',
pape_request = pape.Request(preferred_auth_policies=preferred_auth)
openid_request.addExtension(pape_request)
# Request team info
teams_mapping_auto = getattr(settings, 'OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO', False)
teams_mapping_auto_blacklist = getattr(settings, 'OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO_BLACKLIST', [])
......@@ -257,7 +266,7 @@ def login_complete(request, redirect_field_name=REDIRECT_FIELD_NAME,
try:
user = authenticate(openid_response=openid_response)
except DjangoOpenIDException, e:
return render_failure(request, "Login Failed", exception=e)
return render_failure(request, e.message, exception=e)
if user is not None:
if user.is_active:
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment