allow requiring specific team membership during authentication

if requested team is not found in openid response, deny login.
if requested team is found in openid response, but doesn't map to existing
group, deny login.

this change also includes some refactorings to extract utility functions to
make openid responses to include sreg and teams data.

django_openid_auth/auth.py

@@ -100,6 +100,17 @@ class OpenIDBackend:
             self.update_groups_from_teams(user, teams_response)
             self.update_staff_status_from_teams(user, teams_response)
 
+        teams_required = getattr(settings,
+                                 'OPENID_LAUNCHPAD_TEAMS_REQUIRED', [])
+        if teams_required:
+            teams_mapping = self.get_teams_mapping()
+            groups_required = [group for team, group in teams_mapping.items()
+                               if team in teams_required]
+            matches = set(groups_required).intersection(
+                user.groups.values_list('name', flat=True))
+            if not matches:
+                return None
+
         return user
 
     def _extract_user_details(self, openid_response):
@@ -165,7 +176,7 @@ class OpenIDBackend:
         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'
 
@@ -183,12 +194,12 @@ class OpenIDBackend:
                 user__username__startswith=nickname)
             # No exception means we have an existing user for this identity
             # that starts with this nickname.
-            
+
             # If they are an exact match, the user already exists and hasn't
             # changed their username, so continue to use it
             if nickname == user_openid.user.username:
                 return nickname
-            
+
             # It is possible we've had to assign them to nickname+i already.
             oid_username = user_openid.user.username
             if len(oid_username) > len(nickname):
@@ -289,7 +300,7 @@ class OpenIDBackend:
         if updated:
             user.save()
 
-    def update_groups_from_teams(self, user, teams_response):
+    def get_teams_mapping(self):
         teams_mapping_auto = getattr(settings, 'OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO', False)
         teams_mapping_auto_blacklist = getattr(settings, 'OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO_BLACKLIST', [])
         teams_mapping = getattr(settings, 'OPENID_LAUNCHPAD_TEAMS_MAPPING', {})
@@ -299,7 +310,10 @@ class OpenIDBackend:
             all_groups = Group.objects.exclude(name__in=teams_mapping_auto_blacklist)
             for group in all_groups:
                 teams_mapping[group.name] = group.name
+        return teams_mapping
 
+    def update_groups_from_teams(self, user, teams_response):
+        teams_mapping = self.get_teams_mapping()
         if len(teams_mapping) == 0:
             return
 

django_openid_auth/tests/test_auth.py

@@ -29,10 +29,11 @@
 import unittest
 
 from django.conf import settings
-from django.contrib.auth.models import User
+from django.contrib.auth.models import Group, User
 from django.test import TestCase
 
 from django_openid_auth.auth import OpenIDBackend
+from django_openid_auth.teams import ns_uri as TEAMS_NS
 from openid.consumer.consumer import SuccessResponse
 from openid.consumer.discover import OpenIDServiceEndpoint
 from openid.message import Message, OPENID2_NS
@@ -48,25 +49,58 @@ class OpenIDBackendTests(TestCase):
         self.backend = OpenIDBackend()
         self.old_openid_use_email_for_username = getattr(settings,
             'OPENID_USE_EMAIL_FOR_USERNAME', False)
+        self.old_openid_launchpad_teams_required = getattr(settings,
+            'OPENID_LAUNCHPAD_TEAMS_REQUIRED', [])
+        self.old_openid_launchpad_teams_mapping_auto = getattr(settings,
+            'OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO', False)
 
     def tearDown(self):
         settings.OPENID_USE_EMAIL_FOR_USERNAME = \
             self.old_openid_use_email_for_username
+        settings.OPENID_LAUNCHPAD_TEAMS_REQUIRED = (
+            self.old_openid_launchpad_teams_required)
+        settings.OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO = (
+            self.old_openid_launchpad_teams_mapping_auto)
 
     def test_extract_user_details_sreg(self):
+        expected = {
+            'nickname': 'someuser',
+            'first_name': 'Some',
+            'last_name': 'User',
+            'email': 'foo@example.com',
+        }
+        data = {
+            'nickname': expected['nickname'],
+            'fullname': "%s %s" % (expected['first_name'],
+                                   expected['last_name']),
+            'email': expected['email'],
+        }
+        response = self.make_response_sreg(**data)
+
+        data = self.backend._extract_user_details(response)
+        self.assertEqual(data, expected)
+
+    def make_fake_openid_endpoint(self, claimed_id=None):
         endpoint = OpenIDServiceEndpoint()
+        endpoint.claimed_id = claimed_id
+        return endpoint
+
+    def make_openid_response(self, sreg_args=None, teams_args=None):
+        endpoint = self.make_fake_openid_endpoint(claimed_id='some-id')
         message = Message(OPENID2_NS)
-        message.setArg(SREG_NS, "nickname", "someuser")
-        message.setArg(SREG_NS, "fullname", "Some User")
-        message.setArg(SREG_NS, "email", "foo@example.com")
+        if sreg_args is not None:
+            for key, value in sreg_args.items():
+                message.setArg(SREG_NS, key, value)
+        if teams_args is not None:
+            for key, value in teams_args.items():
+                message.setArg(TEAMS_NS, key, value)
         response = SuccessResponse(
             endpoint, message, signed_fields=message.toPostArgs().keys())
+        return response
 
-        data = self.backend._extract_user_details(response)
-        self.assertEqual(data, {"nickname": "someuser",
-                                "first_name": "Some",
-                                "last_name": "User",
-                                "email": "foo@example.com"})
+    def make_response_sreg(self, **kwargs):
+        response = self.make_openid_response(sreg_args=kwargs)
+        return response
 
     def make_response_ax(self, schema="http://axschema.org/",
         fullname="Some User", nickname="someuser", email="foo@example.com",
@@ -180,6 +214,42 @@ class OpenIDBackendTests(TestCase):
             self.assertEqual(expected,
                 self.backend._get_preferred_username(nick, email))
 
+    def test_authenticate_when_not_member_of_teams_required(self):
+        settings.OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO = True
+        settings.OPENID_LAUNCHPAD_TEAMS_REQUIRED = ['team']
+        Group.objects.create(name='team')
+
+        response = self.make_openid_response(
+            sreg_args=dict(nickname='someuser'),
+            teams_args=dict(is_member='foo'))
+        user = self.backend.authenticate(openid_response=response)
+
+        self.assertIsNone(user)
+
+    def test_authenticate_when_no_group_mapping_to_required_team(self):
+        settings.OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO = True
+        settings.OPENID_LAUNCHPAD_TEAMS_REQUIRED = ['team']
+        assert Group.objects.filter(name='team').count() == 0
+
+        response = self.make_openid_response(
+            sreg_args=dict(nickname='someuser'),
+            teams_args=dict(is_member='foo'))
+        user = self.backend.authenticate(openid_response=response)
+
+        self.assertIsNone(user)
+
+    def test_authenticate_when_member_of_teams_required(self):
+        settings.OPENID_LAUNCHPAD_TEAMS_MAPPING_AUTO = True
+        settings.OPENID_LAUNCHPAD_TEAMS_REQUIRED = ['team']
+        Group.objects.create(name='team')
+
+        response = self.make_openid_response(
+            sreg_args=dict(nickname='someuser'),
+            teams_args=dict(is_member='foo,team'))
+        user = self.backend.authenticate(openid_response=response)
+
+        self.assertIsNotNone(user)
+
 
 def suite():
     return unittest.TestLoader().loadTestsFromName(__name__)