Cleanup exception class heirarchy and imports, make overriding the login failure…

Cleanup exception class heirarchy and imports, make overriding the login failure handler a settings option, add additional test cases to veryfiy that overriding the handler works and that the proper exceptions are being passed

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

@@ -38,7 +38,6 @@ from openid.extensions import ax, sreg, pape
 from django_openid_auth import teams
 from django_openid_auth.models import UserOpenID
 from django_openid_auth.exceptions import (
-    DjangoOpenIDException,
     IdentityAlreadyClaimed,
     DuplicateUsernameViolation,
     MissingUsernameViolation,

django_openid_auth/exceptions.py

@@ -39,10 +39,7 @@ class IdentityAlreadyClaimed(DjangoOpenIDException):
         else:
             self.message = message
 
-class StrictUsernameViolation(DjangoOpenIDException):
-    pass
-    
-class DuplicateUsernameViolation(StrictUsernameViolation):
+class DuplicateUsernameViolation(DjangoOpenIDException):
 
     def __init__(self, message=None):
         if message is None:
@@ -50,7 +47,7 @@ class DuplicateUsernameViolation(StrictUsernameViolation):
         else:
             self.message = message
 
-class MissingUsernameViolation(StrictUsernameViolation):
+class MissingUsernameViolation(DjangoOpenIDException):
 
     def __init__(self, message=None):
         if message is None:

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
@@ -56,11 +56,14 @@ 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,
+)
 
 ET = importElementTree()
 
-
 class StubOpenIDProvider(HTTPFetcher):
 
     def __init__(self, base_url):
@@ -182,6 +185,7 @@ 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
 
 
@@ -204,6 +208,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
 
         setDefaultFetcher(None)
@@ -485,6 +490,64 @@ 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)
+
+    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.assertEquals(message, 'Login Failed')
+           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
@@ -701,6 +764,43 @@ 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, '<p>Login Failed</p>', status_code=403)
+
+    def test_strict_username_no_nickname_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.assertEquals(message, 'Login Failed')
+           self.assertTrue(isinstance(exception, 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
@@ -732,6 +832,50 @@ 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, '<p>Login Failed</p>', 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.assertEquals(message, 'Login Failed')
+           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_update_details(self):
         settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
@@ -1001,7 +1145,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

@@ -55,13 +55,7 @@ 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,
-    IdentityAlreadyClaimed,
-    DuplicateUsernameViolation,
-    MissingUsernameViolation,
-    MissingPhysicalMultiFactor,
-)
+from django_openid_auth.exceptions import DjangoOpenIDException
 
 
 next_url_re = re.compile('^/[-\w/]+$')
@@ -248,8 +242,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: