session limit enforcement

* upon creating a new session, invalidate oldest sessions

awx/main/tests/__init__.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2015 Ansible, Inc.
 # All Rights Reserved.
 
-from awx.main.tests.organizations import OrganizationsTest # noqa
+from awx.main.tests.organizations import * # noqa
 from awx.main.tests.users import * # noqa
 from awx.main.tests.inventory import * # noqa
 from awx.main.tests.projects import ProjectsTest, ProjectUpdatesTest # noqa

awx/main/tests/organizations.py

@@ -1,10 +1,56 @@
 # Copyright (c) 2015 Ansible, Inc.
 # All Rights Reserved.
 
+# Python
+from datetime import timedelta
+
+# Django
 from django.core.urlresolvers import reverse
+from django.test.utils import override_settings
+from django.contrib.auth.models import User
+from django.utils.timezone import now as tz_now
+
+# AWX
 from awx.main.models import * # noqa
 from awx.main.tests.base import BaseTest
 
+__all__ = ['AuthTokenLimitUnitTest', 'OrganizationsTest']
+
+class AuthTokenLimitUnitTest(BaseTest):
+
+    def setUp(self):
+        self.now = tz_now()
+        # Times are relative to now
+        # (key, created on in seconds , expiration in seconds)
+        self.test_data = [
+            # a is implicitly expired
+            ("a", -1000, -10),
+            # b's are invalid due to session limit of 3
+            ("b", -100, 60),
+            ("bb", -100, 60),
+            ("c", -90, 70),
+            ("d", -80, 80),
+            ("e", -70, 90),
+        ]
+        self.user = User.objects.create_superuser('admin', 'foo@bar.com', 'password')
+        for key, t_create, t_expire in self.test_data:
+            AuthToken.objects.create(
+                user=self.user,
+                key=key,
+                request_hash='this_is_a_hash',
+                created=self.now + timedelta(seconds=t_create),
+                expires=self.now + timedelta(seconds=t_expire),
+            )
+        super(AuthTokenLimitUnitTest, self).setUp()
+
+    @override_settings(AUTH_TOKEN_PER_USER=3)
+    def test_get_tokens_over_limit(self):
+        invalid_tokens = AuthToken.get_tokens_over_limit(self.user, now=self.now)
+        invalid_keys = [x.key for x in invalid_tokens]
+        self.assertEqual(len(invalid_keys), 2)
+        self.assertIn('b', invalid_keys)
+        self.assertIn('bb', invalid_keys)
+
 class OrganizationsTest(BaseTest):
 
     def collection(self):

awx/main/tests/users.py

@@ -4,18 +4,20 @@
 # Python
 import datetime
 import urllib
+from mock import patch
 
 # Django
 from django.conf import settings
 from django.contrib.auth.models import User, Group
 from django.db.models import Q
 from django.core.urlresolvers import reverse
+from django.test.utils import override_settings
 
 # AWX
 from awx.main.models import * # noqa
 from awx.main.tests.base import BaseTest
 
-__all__ = ['AuthTokenTimeoutTest', 'AuthTokenProxyTest', 'UsersTest', 'LdapTest']
+__all__ = ['AuthTokenTimeoutTest', 'AuthTokenLimitTest', 'AuthTokenProxyTest', 'UsersTest', 'LdapTest']
 
 
 class AuthTokenTimeoutTest(BaseTest):
@@ -38,6 +40,41 @@ class AuthTokenTimeoutTest(BaseTest):
         self.assertIn('Auth-Token-Timeout', response)
         self.assertEqual(response['Auth-Token-Timeout'], str(settings.AUTH_TOKEN_EXPIRATION))
 
+class AuthTokenLimitTest(BaseTest):
+    def setUp(self):
+        super(AuthTokenLimitTest, self).setUp()
+        self.setup_users()
+        self.setup_instances()
+
+    @override_settings(AUTH_TOKEN_PER_USER=1)
+    @patch.object(awx.main.models.organization.AuthToken, 'get_request_hash')
+    def test_invalidate_first_session(self, mock_get_request_hash):
+        auth_token_url = reverse('api:auth_token_view')
+        user_me_url = reverse('api:user_me_list')
+
+        data = dict(zip(('username', 'password'), self.get_normal_credentials()))
+
+        mock_get_request_hash.return_value = "session_1"
+        response = self.post(auth_token_url, data, expect=200, auth=None)
+        auth_token1 = {
+            'token': response['token']
+        }
+        self.get(user_me_url, expect=200, auth=auth_token1)
+        
+        mock_get_request_hash.return_value = "session_2"
+        response = self.post(auth_token_url, data, expect=200, auth=None)
+        auth_token2 = {
+            'token': response['token']
+        }
+        self.get(user_me_url, expect=200, auth=auth_token2)
+
+        # Ensure our get_request_hash mock is working
+        self.assertNotEqual(auth_token1['token'], auth_token2['token'])
+
+        mock_get_request_hash.return_value = "session_1"
+        response = self.get(user_me_url, expect=401, auth=auth_token1)
+        self.assertEqual(AuthToken.reason_long('limit_reached'), response['detail'])
+
 '''
 Ensure ips from the X-Forwarded-For get honored and used in auth tokens
 '''
@@ -225,7 +262,7 @@ class UsersTest(BaseTest):
         remote_addr = '127.0.0.2'
         response = self.get(user_me_url, expect=401, auth=auth_token,
                             remote_addr=remote_addr)
-        self.assertEqual(response['detail'], 'Invalid token')
+        self.assertEqual(response['detail'], AuthToken.reason_long('invalid_token'))
 
         # The WWW-Authenticate header should specify Token auth, since that
         # auth method was used in the request.