more refactoring and added a unit test

This commit is contained in:
Ken Cochrane 2014-12-31 19:44:21 -05:00
parent 15e6a40e0a
commit 09f29131a4
2 changed files with 47 additions and 32 deletions

View file

@ -11,7 +11,8 @@ from django.core.urlresolvers import NoReverseMatch
from django.core.urlresolvers import reverse from django.core.urlresolvers import reverse
from .utils import ( from .utils import (
COOLOFF_TIME, FAILURE_LIMIT, reset_failed_attempts) COOLOFF_TIME, FAILURE_LIMIT, reset_failed_attempts,
is_valid_ip)
redis_client = mockredis.mock_strict_redis_client() redis_client = mockredis.mock_strict_redis_client()
@ -154,3 +155,16 @@ class AccessAttemptTest(TestCase):
# Make a login attempt again # Make a login attempt again
self.test_valid_login() self.test_valid_login()
def test_is_valid_ip(self):
"""
Test the is_valid_ip() method
"""
self.assertEquals(is_valid_ip('192.168.0.1'), True)
self.assertEquals(is_valid_ip('130.80.100.24'), True)
self.assertEquals(is_valid_ip('8.8.8.8'), True)
self.assertEquals(is_valid_ip('127.0.0.1'), True)
self.assertEquals(is_valid_ip('fish'), False)
self.assertEquals(is_valid_ip(None), False)
self.assertEquals(is_valid_ip(''), False)

View file

@ -11,48 +11,55 @@ from django.utils.translation import ugettext_lazy
from .models import AccessAttempt from .models import AccessAttempt
REDIS_HOST = settings.REDIS_HOST
REDIS_PORT = settings.REDIS_PORT def get_setting(variable, default=None):
REDIS_PASSWORD = settings.REDIS_PASSWORD """ get the 'variable' from settings if not there use the
REDIS_DB = settings.REDIS_DB provided default """
return getattr(settings, variable, default)
# redis server host
REDIS_HOST = get_setting('REDIS_HOST')
# redis server port
REDIS_PORT = get_setting('REDIS_PORT')
# redis server password
REDIS_PASSWORD = get_setting('REDIS_PASSWORD')
# redis db
REDIS_DB = get_setting('REDIS_DB')
# see if the user has overridden the failure limit # see if the user has overridden the failure limit
FAILURE_LIMIT = getattr(settings, 'DEFENDER_LOGIN_FAILURE_LIMIT', 3) FAILURE_LIMIT = get_setting('DEFENDER_LOGIN_FAILURE_LIMIT', 3)
USE_USER_AGENT = getattr(settings, 'DEFENDER_USE_USER_AGENT', False) USE_USER_AGENT = get_setting('DEFENDER_USE_USER_AGENT', False)
# use a specific username field to retrieve from login POST data # use a specific username field to retrieve from login POST data
USERNAME_FORM_FIELD = getattr(settings, USERNAME_FORM_FIELD = get_setting('DEFENDER_USERNAME_FORM_FIELD', 'username')
'DEFENDER_USERNAME_FORM_FIELD',
'username')
# see if the django app is sitting behind a reverse proxy # see if the django app is sitting behind a reverse proxy
BEHIND_REVERSE_PROXY = getattr(settings, BEHIND_REVERSE_PROXY = get_setting('DEFENDER_BEHIND_REVERSE_PROXY', False)
'DEFENDER_BEHIND_REVERSE_PROXY',
False)
# the prefix for these keys in your cache. # the prefix for these keys in your cache.
CACHE_PREFIX = getattr(settings, CACHE_PREFIX = get_setting('DEFENDER_CACHE_PREFIX', 'defender')
'DEFENDER_CACHE_PREFIX',
'defender')
# if the django app is behind a reverse proxy, look for the # if the django app is behind a reverse proxy, look for the
# ip address using this HTTP header value # ip address using this HTTP header value
REVERSE_PROXY_HEADER = getattr(settings, REVERSE_PROXY_HEADER = get_setting('DEFENDER_REVERSE_PROXY_HEADER',
'DEFENDER_REVERSE_PROXY_HEADER', 'HTTP_X_FORWARDED_FOR')
'HTTP_X_FORWARDED_FOR')
# how long to wait before the bad login attempt gets forgotten. in seconds. # how long to wait before the bad login attempt gets forgotten. in seconds.
COOLOFF_TIME = getattr(settings, 'DEFENDER_COOLOFF_TIME', 300) # seconds COOLOFF_TIME = get_setting('DEFENDER_COOLOFF_TIME', 300) # seconds
LOCKOUT_TEMPLATE = getattr(settings, 'DEFENDER_LOCKOUT_TEMPLATE', None) LOCKOUT_TEMPLATE = get_setting('DEFENDER_LOCKOUT_TEMPLATE')
ERROR_MESSAGE = ugettext_lazy("Please enter a correct username and password. " ERROR_MESSAGE = ugettext_lazy("Please enter a correct username and password. "
"Note that both fields are case-sensitive.") "Note that both fields are case-sensitive.")
# use a specific username field to retrieve from login POST data # use a specific username field to retrieve from login POST data
USERNAME_FORM_FIELD = getattr(settings, USERNAME_FORM_FIELD = get_setting('DEFENDER_USERNAME_FORM_FIELD', 'username')
'DEFENDER_USERNAME_FORM_FIELD',
'username') LOCKOUT_URL = get_setting('DEFENDER_LOCKOUT_URL')
redis_server = redis.StrictRedis( redis_server = redis.StrictRedis(
host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, password=REDIS_PASSWORD) host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, password=REDIS_PASSWORD)
@ -122,11 +129,6 @@ def get_ip(request):
return ip return ip
def get_lockout_url():
""" get the lockout url from the settings """
return getattr(settings, 'DEFENDER_LOCKOUT_URL', None)
def get_ip_attempt_cache_key(ip): def get_ip_attempt_cache_key(ip):
""" get the cache key by ip """ """ get the cache key by ip """
return "{0}:failed:ip:{1}".format(CACHE_PREFIX, ip) return "{0}:failed:ip:{1}".format(CACHE_PREFIX, ip)
@ -156,7 +158,7 @@ def increment_key(key):
def get_user_attempts(request): def get_user_attempts(request):
"""Returns number of access attempts for this ip, username """ Returns number of access attempts for this ip, username
""" """
ip = get_ip(request) ip = get_ip(request)
@ -177,7 +179,7 @@ def get_user_attempts(request):
def block_ip(ip): def block_ip(ip):
""" given the ip, block it""" """ given the ip, block it """
key = get_ip_blocked_cache_key(ip) key = get_ip_blocked_cache_key(ip)
redis_server.set(key, 'blocked', COOLOFF_TIME) redis_server.set(key, 'blocked', COOLOFF_TIME)
@ -225,7 +227,6 @@ def lockout_response(request):
return render_to_response(LOCKOUT_TEMPLATE, context, return render_to_response(LOCKOUT_TEMPLATE, context,
context_instance=RequestContext(request)) context_instance=RequestContext(request))
LOCKOUT_URL = get_lockout_url()
if LOCKOUT_URL: if LOCKOUT_URL:
return HttpResponseRedirect(LOCKOUT_URL) return HttpResponseRedirect(LOCKOUT_URL)