Merged in lp:~mhall119/django-openid-auth/provides-784239

README.txt

@@ -154,3 +154,15 @@ If your users should use a physical multi-factor authentication method, such as
 If the user's OpenID provider supports the PAPE extension and provides the Physical Multifactor authentication policy, this will
 cause the OpenID login to fail if the user does not provide valid physical authentication to the provider.
 
+== Override Login Failure Handling ==
+
+You can optionally provide your own handler for login failures by adding the following setting:
+
+        OPENID_RENDER_FAILURE = failure_handler_function
+
+Where failure_handler_function is a function reference that will take the following parameters:
+
+        def failure_handler_function(request, message, status=None, template_name=None, exception=None)
+
+This function must return a Django.http.HttpResponse instance.
+

django_openid_auth/auth.py

@@ -37,16 +37,13 @@ from openid.extensions import ax, sreg, pape
 
 from django_openid_auth import teams
 from django_openid_auth.models import UserOpenID
-
-
-class IdentityAlreadyClaimed(Exception):
-    pass
-
-class StrictUsernameViolation(Exception):
-    pass
-
-class RequiredAttributeNotReturned(Exception):
-    pass
+from django_openid_auth.exceptions import (
+    IdentityAlreadyClaimed,
+    DuplicateUsernameViolation,
+    MissingUsernameViolation,
+    MissingPhysicalMultiFactor,
+    RequiredAttributeNotReturned,
+)
 
 class OpenIDBackend:
     """A django.contrib.auth backend that authenticates the user based on
@@ -92,7 +89,7 @@ class OpenIDBackend:
             pape_response = pape.Response.fromSuccessResponse(openid_response)
             if pape_response is None or \
                pape.AUTH_MULTI_FACTOR_PHYSICAL not in pape_response.auth_policies:
-                return None
+                raise MissingPhysicalMultiFactor()
 
         teams_response = teams.TeamsResponse.fromSuccessResponse(
             openid_response)
@@ -150,6 +147,12 @@ 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 MissingUsernameViolation()
+                
         # If we don't have a nickname, and we're not being strict, use a default
         nickname = nickname or 'openiduser'
 
@@ -184,7 +187,7 @@ class OpenIDBackend:
 
         if getattr(settings, 'OPENID_STRICT_USERNAMES', False):
             if User.objects.filter(username__exact=nickname).count() > 0:
-                raise StrictUsernameViolation(
+                raise DuplicateUsernameViolation(
                     "The username (%s) with which you tried to log in is "
                     "already in use for a different account." % nickname)
 

django_openid_auth/exceptions.py

+# django-openid-auth -  OpenID integration for django.contrib.auth
+#
+# Copyright (C) 2008-2010 Canonical Ltd.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#
+# * Redistributions in binary form must reproduce the above copyright
+# notice, this list of conditions and the following disclaimer in the
+# documentation and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+"""Exception classes thrown by OpenID Authentication and Validation."""
+
+class DjangoOpenIDException(Exception):
+    pass
+
+class RequiredAttributeNotReturned(DjangoOpenIDException):
+    pass
+
+class IdentityAlreadyClaimed(DjangoOpenIDException):
+
+    def __init__(self, message=None):
+        if message is None:
+            self.message = "Another user already exists for your selected OpenID"
+        else:
+            self.message = message
+
+class DuplicateUsernameViolation(DjangoOpenIDException):
+
+    def __init__(self, message=None):
+        if message is None:
+            self.message = "Your desired username is already being used."
+        else:
+            self.message = message
+
+class MissingUsernameViolation(DjangoOpenIDException):
+
+    def __init__(self, message=None):
+        if message is None:
+            self.message = "No nickname given for your account."
+        else:
+            self.message = message
+
+class MissingPhysicalMultiFactor(DjangoOpenIDException):
+
+    def __init__(self, message=None):
+        if message is None:
+            self.message = "Login requires physical multi-factor authentication."
+        else:
+            self.message = message
+

django_openid_auth/tests/test_views.py

@@ -32,7 +32,7 @@ from urllib import quote_plus
 
 from django.conf import settings
 from django.contrib.auth.models import User, Group
-from django.http import HttpRequest
+from django.http import HttpRequest, HttpResponse
 from django.test import TestCase
 from openid.consumer.consumer import Consumer, SuccessResponse
 from openid.consumer.discover import OpenIDServiceEndpoint
@@ -53,11 +53,15 @@ from django_openid_auth.views import (
 from django_openid_auth.auth import OpenIDBackend
 from django_openid_auth.signals import openid_login_complete
 from django_openid_auth.store import DjangoOpenIDStore
-
+from django_openid_auth.exceptions import (
+    MissingUsernameViolation,
+    DuplicateUsernameViolation,
+    MissingPhysicalMultiFactor,
+    RequiredAttributeNotReturned,
+)
 
 ET = importElementTree()
 
-
 class StubOpenIDProvider(HTTPFetcher):
 
     def __init__(self, base_url):
@@ -179,7 +183,9 @@ class RelyingPartyTests(TestCase):
         self.old_use_as_admin_login = getattr(settings, 'OPENID_USE_AS_ADMIN_LOGIN', False)
         self.old_follow_renames = getattr(settings, 'OPENID_FOLLOW_RENAMES', False)
         self.old_physical_multifactor = getattr(settings, 'OPENID_PHYSICAL_MULTIFACTOR_REQUIRED', False)
+        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', [])
 
@@ -203,6 +209,7 @@ class RelyingPartyTests(TestCase):
         settings.OPENID_USE_AS_ADMIN_LOGIN = self.old_use_as_admin_login
         settings.OPENID_FOLLOW_RENAMES = self.old_follow_renames
         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
 
@@ -485,6 +492,63 @@ 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 requires physical multi-factor authentication.</p>', status_code=403)
+
+    def test_login_physical_multifactor_not_provided_override(self):
+        settings.OPENID_PHYSICAL_MULTIFACTOR_REQUIRED = True
+        preferred_auth = pape.AUTH_MULTI_FACTOR_PHYSICAL
+        self.provider.type_uris.append(pape.ns_uri)
+
+        # Override the login_failure handler
+        def mock_login_failure_handler(request, message, status=403,
+                                       template_name=None,
+                                       exception=None):
+           self.assertTrue(isinstance(exception, MissingPhysicalMultiFactor))
+           return HttpResponse('Test Failure Override', status=200)
+        settings.OPENID_RENDER_FAILURE = mock_login_failure_handler
+
+        def mock_complete(this, request_args, return_to):
+            request = {'openid.mode': 'checkid_setup',
+                       'openid.trust_root': 'http://localhost/',
+                       'openid.return_to': 'http://localhost/',
+                       'openid.identity': IDENTIFIER_SELECT,
+                       'openid.ns.pape' : pape.ns_uri,
+                       'openid.pape.auth_policies': request_args.get('openid.pape.auth_policies', pape.AUTH_NONE),
+            }
+            openid_server = self.provider.server
+            orequest = openid_server.decodeRequest(request)
+            response = SuccessResponse(
+                self.endpoint, orequest.message,
+                signed_fields=['openid.pape.auth_policies',])
+            return response
+        Consumer.complete = mock_complete
+
+        user = User.objects.create_user('testuser', 'test@example.com')
+        useropenid = UserOpenID(    
+            user=user,
+            claimed_id='http://example.com/identity',
+            display_id='http://example.com/identity')
+        useropenid.save()
+
+        openid_req = {'openid_identifier': 'http://example.com/identity',
+               'next': '/getuser/'}
+        openid_resp =  {'nickname': 'testuser', 'fullname': 'Openid User',
+                 'email': 'test@example.com'}
+
+        openid_request = self._get_login_request(openid_req)
+        openid_response = self._get_login_response(openid_request, openid_req, openid_resp, use_pape=pape.AUTH_NONE)
+
+        response_auth = openid_request.message.getArg(
+            'http://specs.openid.net/extensions/pape/1.0',
+            'auth_policies',
+        )
+        self.assertNotEqual(response_auth, preferred_auth)
+
+        # Status code should be 200, since we over-rode the login_failure handler
+        response = self.complete(openid_response)
+        self.assertEquals(200, response.status_code)
+        self.assertContains(response, 'Test Failure Override')
 
     def test_login_without_nickname(self):
         settings.OPENID_CREATE_USERS = True
@@ -680,6 +744,7 @@ class RelyingPartyTests(TestCase):
     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/',
@@ -701,6 +766,44 @@ class RelyingPartyTests(TestCase):
 
         # 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, "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.assertTrue(isinstance(exception, (RequiredAttributeNotReturned, MissingUsernameViolation)))
+           return HttpResponse('Test Failure Override', status=200)
+        settings.OPENID_RENDER_FAILURE = mock_login_failure_handler
+        
+        # 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': '', # No nickname
+                           'fullname': 'Some User',
+                           'email': 'foo@example.com'})
+        openid_response.addExtension(sreg_response)
+        response = self.complete(openid_response)
+            
+        # Status code should be 200, since we over-rode the login_failure handler
+        self.assertEquals(200, response.status_code)
+        self.assertContains(response, 'Test Failure Override')
 
     def test_strict_username_duplicate_user(self):
         settings.OPENID_CREATE_USERS = True
@@ -731,11 +834,54 @@ class RelyingPartyTests(TestCase):
         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,
             "The username (someuser) with which you tried to log in is "
             "already in use for a different account.",
             status_code=403)
 
+    def test_strict_username_duplicate_user_override(self):
+        settings.OPENID_CREATE_USERS = True
+        settings.OPENID_STRICT_USERNAMES = True
+
+        # Override the login_failure handler
+        def mock_login_failure_handler(request, message, status=403,
+                                       template_name=None,
+                                       exception=None):
+           self.assertTrue(isinstance(exception, DuplicateUsernameViolation))
+           return HttpResponse('Test Failure Override', status=200)
+        settings.OPENID_RENDER_FAILURE = mock_login_failure_handler
+
+        # Create a user with the same name as we'll pass back via sreg.
+        user = User.objects.create_user('someuser', 'someone@example.com')
+        useropenid = UserOpenID(
+            user=user,
+            claimed_id='http://example.com/different_identity',
+            display_id='http://example.com/different_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/'})
+        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': 'someuser', 'fullname': 'Some User',
+                           'email': 'foo@example.com'})
+        openid_response.addExtension(sreg_response)
+        response = self.complete(openid_response)
+        
+        # Status code should be 200, since we over-rode the login_failure handler
+        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.
@@ -1054,7 +1200,7 @@ class RelyingPartyTests(TestCase):
         self.assertTrue(self.signal_handler_called)
         openid_login_complete.disconnect(login_callback)
 
-
+    
 class HelperFunctionsTest(TestCase):
     def test_sanitise_redirect_url(self):
         settings.ALLOWED_EXTERNAL_OPENID_REDIRECT_DOMAINS = [

django_openid_auth/views.py

@@ -51,14 +51,14 @@ 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
 from django_openid_auth.store import DjangoOpenIDStore
+from django_openid_auth.exceptions import (
+    RequiredAttributeNotReturned,
+    DjangoOpenIDException,
+)
 
 
 next_url_re = re.compile('^/[-\w/]+$')
@@ -122,10 +122,11 @@ def render_openid_request(request, openid_request, return_to, trust_root=None):
 
 
 def default_render_failure(request, message, status=403,
-                           template_name='openid/failure.html'):
+                           template_name='openid/failure.html',
+                           exception=None):
     """Render an error page to the user."""
     data = render_to_string(
-        template_name, dict(message=message),
+        template_name, dict(message=message, exception=exception),
         context_instance=RequestContext(request))
     return HttpResponse(data, status=status)
 
@@ -175,7 +176,8 @@ def login_begin(request, template_name='openid/login.html',
         openid_request = consumer.begin(openid_url)
     except DiscoveryFailure, exc:
         return render_failure(
-            request, "OpenID discovery error: %s" % (str(exc),), status=500)
+            request, "OpenID discovery error: %s" % (str(exc),), status=500,
+            exception=exc)
 
     # Request some user details.  If the provider advertises support
     # for attribute exchange, use that.
@@ -220,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', [])
@@ -250,8 +251,11 @@ def login_begin(request, template_name='openid/login.html',
 
 @csrf_exempt
 def login_complete(request, redirect_field_name=REDIRECT_FIELD_NAME,
-                   render_failure=default_render_failure):
+                   render_failure=None):
     redirect_to = request.REQUEST.get(redirect_field_name, '')
+    render_failure = render_failure or \
+                     getattr(settings, 'OPENID_RENDER_FAILURE', None) or \
+                     default_render_failure
 
     openid_response = parse_openid_response(request)
     if not openid_response:
@@ -261,9 +265,9 @@ def login_complete(request, redirect_field_name=REDIRECT_FIELD_NAME,
     if openid_response.status == SUCCESS:
         try:
             user = authenticate(openid_response=openid_response)
-        except (StrictUsernameViolation, RequiredAttributeNotReturned), e:
-            return render_failure(request, e)
-
+        except DjangoOpenIDException, e:
+            return render_failure(request, e.message, exception=e)
+            
         if user is not None:
             if user.is_active:
                 auth_login(request, user)