Finish test cases, add implementation

django_openid_auth/auth.py

@@ -33,7 +33,7 @@ __metaclass__ = type
 from django.conf import settings
 from django.contrib.auth.models import User, Group
 from openid.consumer.consumer import SUCCESS
-from openid.extensions import ax, sreg
+from openid.extensions import ax, sreg, pape
 
 from django_openid_auth import teams
 from django_openid_auth.models import UserOpenID
@@ -88,6 +88,12 @@ class OpenIDBackend:
             details = self._extract_user_details(openid_response)
             self.update_user_details(user, details, openid_response)
 
+        if getattr(settings, 'OPENID_PHYSICAL_MULTIFACTOR_REQUIRED', False):
+            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
+
         teams_response = teams.TeamsResponse.fromSuccessResponse(
             openid_response)
         if teams_response:

django_openid_auth/tests/__init__.py

@@ -27,6 +27,9 @@
 # POSSIBILITY OF SUCH DAMAGE.
 
 import unittest
+from test_views import *
+from test_store import *
+from test_auth import *
 
 
 def suite():

django_openid_auth/tests/test_views.py

@@ -28,12 +28,14 @@
 
 import cgi
 import unittest
+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.test import TestCase
 from openid.consumer.consumer import Consumer, SuccessResponse
+from openid.consumer.discover import OpenIDServiceEndpoint
 from openid.extensions import ax, sreg, pape
 from openid.fetchers import (
     HTTPFetcher, HTTPFetchingError, HTTPResponse, setDefaultFetcher)
@@ -49,7 +51,9 @@ from django_openid_auth.views import (
     make_consumer,
     login_begin,
     login_complete,
+    parse_openid_response,
 )
+from django_openid_auth.auth import OpenIDBackend
 from django_openid_auth.signals import openid_login_complete
 from django_openid_auth.store import DjangoOpenIDStore
 
@@ -162,6 +166,9 @@ class RelyingPartyTests(TestCase):
         super(RelyingPartyTests, self).setUp()
         self.provider = StubOpenIDProvider('http://example.com/')
         self.req = DummyDjangoRequest('http://localhost/')
+        self.endpoint = OpenIDServiceEndpoint()
+        self.endpoint.claimed_id = 'http://example.com/identity'
+        self.endpoint.server_url = 'http://example.com/'
         self.consumer = make_consumer(self.req)
         self.server = Server(DjangoOpenIDStore())
         setDefaultFetcher(self.provider, wrap_exceptions=False)
@@ -175,6 +182,8 @@ 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_consumer_complete = Consumer.complete
+
 
         settings.OPENID_CREATE_USERS = False
         settings.OPENID_STRICT_USERNAMES = False
@@ -195,6 +204,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
+        Consumer.complete = self.old_consumer_complete
 
         setDefaultFetcher(None)
         super(RelyingPartyTests, self).tearDown()
@@ -332,7 +342,7 @@ class RelyingPartyTests(TestCase):
         self.assertEquals(user.last_name, 'User')
         self.assertEquals(user.email, 'foo@example.com')
 
-    def _do_user_login(self, req_data, resp_data, use_sreg=True, use_pape=False):
+    def _do_user_login(self, req_data, resp_data, use_sreg=True, use_pape=None):
         openid_request = self._get_login_request(req_data)
         openid_response = self._get_login_response(openid_request, resp_data, use_sreg, use_pape)
         response = self.complete(openid_response)
@@ -357,17 +367,15 @@ class RelyingPartyTests(TestCase):
             sreg_response = sreg.SRegResponse.extractResponse(
                 sreg_request, resp_data)
             openid_response.addExtension(sreg_response)
-        if use_pape:
+        if use_pape is not None:
             policies = [
-                pape.AUTH_MULTI_FACTOR_PHYSICAL,
+                use_pape
             ]
             pape_response = pape.Response(auth_policies=policies)
             openid_response.addExtension(pape_response)
         return openid_response
 
-    def get_query(self, response):
-        query_start = response['Location'].find('?')
-        query_str = response['Location'][query_start+1:]
+    def parse_query_string(self, query_str):
         query_items = map(tuple,
             [item.split('=') for item in query_str.split('&')])
         query = dict(query_items)
@@ -394,6 +402,22 @@ class RelyingPartyTests(TestCase):
         preferred_auth = pape.AUTH_MULTI_FACTOR_PHYSICAL
         self.provider.type_uris.append(pape.ns_uri)
 
+        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,
@@ -406,25 +430,40 @@ class RelyingPartyTests(TestCase):
         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=True)
+        response = self._do_user_login(openid_req, openid_resp, use_pape=pape.AUTH_MULTI_FACTOR_PHYSICAL)
+
+        query = self.parse_query_string(response.request['QUERY_STRING'])
+        self.assertTrue('openid.pape.auth_policies' in query)
+        self.assertEqual(query['openid.pape.auth_policies'], 
+                quote_plus(preferred_auth))
+
+        response = self.client.get('/getuser/')
+        self.assertEqual(response.content, 'testuser')
 
-        response_auth = openid_request.message.getArg(
-            'http://specs.openid.net/extensions/pape/1.0',
-            'auth_policies',
-        )
-        self.assertEqual(response_auth, preferred_auth)
 
-        response = self.complete(openid_response)
-        self.assertRedirects(response, 'http://testserver/getuser/')
-    
     def test_login_physical_multifactor_not_provided(self):
         settings.OPENID_PHYSICAL_MULTIFACTOR_REQUIRED = True
         preferred_auth = pape.AUTH_MULTI_FACTOR_PHYSICAL
         self.provider.type_uris.append(pape.ns_uri)
 
+        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(
+        useropenid = UserOpenID(    
             user=user,
             claimed_id='http://example.com/identity',
             display_id='http://example.com/identity')
@@ -436,7 +475,7 @@ class RelyingPartyTests(TestCase):
                  '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=False)
+        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',

django_openid_auth/views.py

@@ -48,7 +48,7 @@ except ImportError:
 from openid.consumer.consumer import (
     Consumer, SUCCESS, CANCEL, FAILURE)
 from openid.consumer.discover import DiscoveryFailure
-from openid.extensions import sreg, ax
+from openid.extensions import sreg, ax, pape
 
 from django_openid_auth import teams
 from django_openid_auth.forms import OpenIDLoginForm
@@ -201,6 +201,14 @@ def login_begin(request, template_name='openid/login.html',
         sreg_optional_fields.extend(extra_fields)
         openid_request.addExtension(
             sreg.SRegRequest(optional=sreg_optional_fields))
+            
+    if getattr(settings, 'OPENID_PHYSICAL_MULTIFACTOR_REQUIRED', False):
+        preferred_auth = [
+            pape.AUTH_MULTI_FACTOR_PHYSICAL,
+        ]
+        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)