Skip to content

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

Merge from trunk

parents b7f4b8e8 fe78850c
Showing with 156 additions and 55 deletions
django_openid_auth/auth.py
......@@ -45,6 +45,9 @@ class IdentityAlreadyClaimed(Exception):
class StrictUsernameViolation(Exception):
pass
class RequiredAttributeNotReturned(Exception):
pass
class OpenIDBackend:
"""A django.contrib.auth backend that authenticates the user based on
an OpenID response."""
......@@ -74,10 +77,7 @@ class OpenIDBackend:
claimed_id__exact=openid_response.identity_url)
except UserOpenID.DoesNotExist:
if getattr(settings, 'OPENID_CREATE_USERS', False):
try:
user = self.create_user_from_openid(openid_response)
except StrictUsernameViolation:
return None
else:
user = user_openid.user
......@@ -138,8 +138,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
......@@ -148,12 +150,6 @@ class OpenIDBackend:
first_name=first_name, last_name=last_name)
def _get_available_username(self, nickname, identity_url):
# If we're being strict about usernames, throw an error if we didn't
# get one back from the provider
if getattr(settings, 'OPENID_STRICT_USERNAMES', False):
if nickname is None or nickname == '':
raise StrictUsernameViolation("No username")
# If we don't have a nickname, and we're not being strict, use a default
nickname = nickname or 'openiduser'
......@@ -188,7 +184,9 @@ class OpenIDBackend:
if getattr(settings, 'OPENID_STRICT_USERNAMES', False):
if User.objects.filter(username__exact=nickname).count() > 0:
raise StrictUsernameViolation("Duplicate username: %s" % nickname)
raise StrictUsernameViolation(
"The username (%s) with which you tried to log in is "
"already in use for a different account." % nickname)
# Pick a username for the user based on their nickname,
# checking for conflicts.
......@@ -206,6 +204,16 @@ class OpenIDBackend:
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 ''
......@@ -240,10 +248,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/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
# Include fullname 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())
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
......@@ -180,7 +180,8 @@ class RelyingPartyTests(TestCase):
self.old_follow_renames = getattr(settings, 'OPENID_FOLLOW_RENAMES', False)
self.old_physical_multifactor = getattr(settings, 'OPENID_PHYSICAL_MULTIFACTOR_REQUIRED', False)
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
......@@ -190,6 +191,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
......@@ -202,6 +204,7 @@ class RelyingPartyTests(TestCase):
settings.OPENID_FOLLOW_RENAMES = self.old_follow_renames
settings.OPENID_PHYSICAL_MULTIFACTOR_REQUIRED = self.old_physical_multifactor
Consumer.complete = self.old_consumer_complete
settings.OPENID_SREG_REQUIRED_FIELDS = self.old_required_fields
setDefaultFetcher(None)
super(RelyingPartyTests, self).tearDown()
......@@ -728,7 +731,39 @@ class RelyingPartyTests(TestCase):
response = self.complete(openid_response)
# Status code should be 403: Forbidden
self.assertEquals(403, response.status_code)
self.assertContains(response,
"The username (someuser) with which you tried to log in is "
"already in use for a different account.",
status_code=403)
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
......@@ -774,6 +809,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
......@@ -51,6 +51,10 @@ from openid.consumer.discover import DiscoveryFailure
from openid.extensions import sreg, ax, pape
from django_openid_auth import teams
from django_openid_auth.auth import (
RequiredAttributeNotReturned,
StrictUsernameViolation,
)
from django_openid_auth.forms import OpenIDLoginForm
from django_openid_auth.models import UserOpenID
from django_openid_auth.signals import openid_login_complete
......@@ -196,11 +200,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 = [
......@@ -248,7 +259,11 @@ def login_complete(request, redirect_field_name=REDIRECT_FIELD_NAME,
request, 'This is an OpenID relying party endpoint.')
if openid_response.status == SUCCESS:
try:
user = authenticate(openid_response=openid_response)
except (StrictUsernameViolation, RequiredAttributeNotReturned), e:
return render_failure(request, e)
if user is not None:
if user.is_active:
auth_login(request, user)
......
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