Merge pull request #61 from 0x836/add_disable_username_block_feature

Add disable username block feature

README.md

@@ -310,6 +310,7 @@ record is created for the failed logins.  [Default: ``3``]
 [Default: ``False``]
 * ``DEFENDER_LOCK_OUT_BY_IP_AND_USERNAME``: Boolean: Locks a user out based on a combination of IP and Username.  This stops a user denying access to the application for all other users accessing the app from behind the same IP address. [Default: ``False``]
 * ``DEFENDER_DISABLE_IP_LOCKOUT``: Boolean: If this is True, it will not lockout the users IP address, it will only lockout the username. [Default: False]
+* ``DEFENDER_DISABLE_USERNAME_LOCKOUT``: Boolean: If this is True, it will not lockout usernames, it will only lockout IP addresess. [Default: False]
 * ``DEFENDER_COOLOFF_TIME``: Int: If set, defines a period of inactivity after which
 old failed login attempts will be forgotten. An integer, will be interpreted as a
 number of seconds. If ``0``, the locks will not expire. [Default: ``300``]

defender/config.py

@@ -24,6 +24,10 @@ LOCKOUT_BY_IP_USERNAME = get_setting(
 # there are too many login attempts.
 DISABLE_IP_LOCKOUT = get_setting('DEFENDER_DISABLE_IP_LOCKOUT', False)
 
+# If this is True, usernames will not get locked when
+# there are too many login attempts.
+DISABLE_USERNAME_LOCKOUT = get_setting('DEFENDER_DISABLE_USERNAME_LOCKOUT', False)
+
 # use a specific username field to retrieve from login POST data
 USERNAME_FORM_FIELD = get_setting('DEFENDER_USERNAME_FORM_FIELD', 'username')
 

defender/tests.py

@@ -666,6 +666,54 @@ class AccessAttemptTest(DefenderTestCase):
         data_out = utils.get_blocked_ips()
         self.assertEqual(data_out, [])
 
+    @patch('defender.config.DISABLE_USERNAME_LOCKOUT', True)
+    def test_disable_username_lockout(self):
+        """Check lockouting still works when we disable username lockout"""
+
+        username = 'testy'
+
+        # try logging in with the same username, but different IPs.
+        # we shouldn't be locked.
+        for i in range(0, config.FAILURE_LIMIT+10):
+            ip = '74.125.126.{0}'.format(i)
+            response = self._login(username=username, remote_addr=ip)
+            # Check if we are in the same login page
+            self.assertContains(response, LOGIN_FORM_KEY)
+
+        # same ip and same username
+        ip = '74.125.127.1'
+        for i in range(0, config.FAILURE_LIMIT):
+            response = self._login(username=username, remote_addr=ip)
+            # Check if we are in the same login page
+            self.assertContains(response, LOGIN_FORM_KEY)
+
+        # But we should get one now
+        # same username and Ip, over the limit.
+        response = self._login(username=username, remote_addr=ip)
+        self.assertContains(response, self.LOCKED_MESSAGE)
+
+        # We shouldn't get a lockout message when attempting to use no username
+        response = self.client.get(ADMIN_LOGIN_URL)
+        self.assertContains(response, LOGIN_FORM_KEY)
+
+        # We shouldn't get a lockout message when attempting to use a different ip address
+        # to be sure that username is not blocked.
+        second_ip = '74.125.127.2'
+        response = self._login(username=username, remote_addr=second_ip)
+        # Check if we are in the same login page
+        self.assertContains(response, LOGIN_FORM_KEY)
+
+        # we should have no usernames are blocked
+        data_out = utils.get_blocked_usernames()
+        self.assertEqual(data_out, [])
+
+        # even if we try to manually block one it still won't be in there.
+        utils.block_username(username)
+
+        # we should still have no ip's blocked
+        data_out = utils.get_blocked_usernames()
+        self.assertEqual(data_out, [])
+
 
 class DefenderTestCaseTest(DefenderTestCase):
     """Make sure that we're cleaning the cache between tests"""

defender/utils.py

@@ -97,6 +97,9 @@ def get_blocked_ips():
 
 def get_blocked_usernames():
     """ get a list of blocked usernames from redis """
+    if config.DISABLE_USERNAME_LOCKOUT:
+        # There are no blocked usernames since we disabled them.
+        return []
     key = get_username_blocked_cache_key("*")
     key_list = [redis_key.decode('utf-8')
                 for redis_key in REDIS_SERVER.keys(key)]
@@ -156,6 +159,9 @@ def block_username(username):
     if not username:
         # no reason to continue when there is no username
         return
+    if config.DISABLE_USERNAME_LOCKOUT:
+        # no need to block, we disabled it.
+        return
     key = get_username_blocked_cache_key(username)
     if config.COOLOFF_TIME:
         REDIS_SERVER.set(key, 'blocked', config.COOLOFF_TIME)
@@ -177,12 +183,12 @@ def record_failed_attempt(ip_address, username):
             ip_block = True
 
     user_block = False
-    user_count = increment_key(get_username_attempt_cache_key(username))
-
-    # if over the limit, add to block
-    if user_count > config.FAILURE_LIMIT:
-        block_username(username)
-        user_block = True
+    if not config.DISABLE_USERNAME_LOCKOUT:
+        user_count = increment_key(get_username_attempt_cache_key(username))
+        # if over the limit, add to block
+        if user_count > config.FAILURE_LIMIT:
+            block_username(username)
+            user_block = True
 
     # if we have this turned on, then there is no reason to look at ip_block
     # we will just look at user_block, and short circut the result since
@@ -192,6 +198,10 @@ def record_failed_attempt(ip_address, username):
         # we need to return False
         return not user_block
 
+    if config.DISABLE_USERNAME_LOCKOUT:
+        # The same as DISABLE_IP_LOCKOUT
+        return not ip_block
+
     # we want to make sure both the IP and user is blocked before we
     # return False
     # this is mostly used when a lot of your users are using proxies,
@@ -267,6 +277,8 @@ def is_user_already_locked(username):
     """Is this username already locked?"""
     if username is None:
         return False
+    if config.DISABLE_USERNAME_LOCKOUT:
+        return False
     return REDIS_SERVER.get(get_username_blocked_cache_key(username))