Preventing login hacks: Rate limiting using memcached

One of the oldest tricks in the book to hack a system is to launch a dictionary attack against a system, i.e. you have a dictionary of words and then your try to login with these. Tons of users use simple passwords, so these kind of attacks are rather effective. In January 2009 Twitter got hacked by such an attack, so a lot of systems can still be hacked by a little script and some good old brute-forcing.

Preventing such an attack can be done by rate limiting logins - e.g. a normal user would not try to login 50 times in 3 minutes, if you observe this kind of behavior then it's likely to be a hacker! A simple Python implementation of a rate limiter was shown by Simon Willson (co-creator of Django):

If you don't have a rate limiter then you are vulnerable for dictionary attacks and I would urge you to use one.

I don't code in Django, so I could not use Simon's code directly... I was therefor forced to reimplement. My modifications include:

  • the rate limiter is not tied up to Django. Only requirement is a memcache client
  • Simon only limits IP's, while my extension can rate limit such things like usernames and passwords (to improve the overall security)

You can use this code freely: 

import types
from datetime import datetime, timedelta
from parts.cache import get_cache

class RateLimiter(object):

    minutes = 3 # The time period
    prefix = 'rl-' # Prefix for memcache key

    expire_after = (minutes + 1) * 60

    def __init__(self, **options):
        for key, value in options.items():
            setattr(self, key, value)

    def register(self, limit_key, requests=25):
        """Register `limit_key`, if the current count
        is greater than `requests` then a LimitExceeded
        error is thrown.
        """
        self._cache_incr( self._current_key(limit_key) )

        if self.get_current_count(limit_key) >= requests:
            raise LimitExceeded()

    def get_current_count(self, limit_key):
        """Return the aggregated count for `limit_key`.
        """
        counts = self._get_counters(limit_key).values()
        return sum(counts)

    def keys_to_check(self, limit_key):
        """Returns the memcached keys for `limit_key`.
        """
        now = datetime.now()
        return [
            '%s%s-%s' % (
                self.prefix,
                limit_key,
                (now - timedelta(minutes = minute)).strftime('%Y%m%d%H%M')
            ) for minute in range(self.minutes + 1)
        ]

    def reset_counts(self, limit_key):
        """Reset the counter for `limit_key`.
        """
        get_cache().delete_multi( self.keys_to_check(limit_key) )


    #--- Private ----------------------------------------------
    def _get_counters(self, limit_key):
        counts = get_cache().get_multi(self.keys_to_check(limit_key))
        for k, v in counts.items():
            counts[k] = long(v)
        return counts

    def _cache_incr(self, key):
        cur_cache = get_cache()
        try:
            cur_cache.add(key, '0', time=self.expire_after)
            cur_cache.incr(key)
        except AttributeError:
            cur = cur_cache.get(key)
            cur = long(cur) if cur else 0
            cur_cache.set(key, cur + 1, self.expire_after)

    def _current_key(self, limit_key):
        return '%s%s-%s' % (
            self.prefix,
            limit_key,
            datetime.now().strftime('%Y%m%d%H%M')
        )

rate_limiter = RateLimiter()

class LimitExceeded(Exception):
    pass

Note that get_cache should be a function that returns a memcache.Client instance.

You can use it in a following way: 

from rate_limit import rate_limiter, LimitExceeded

rate_limit_keys = [
    request().headers.get('X-Real-IP', ''),
    generate_md5(nick_name.encode('utf8')),
]

try:
    for k in rate_limit_keys:
        rate_limiter.register(k)
except LimitExceeded:
    #... handle error
5. Apr 2009 • Code · Python · Security · Tips
© Amir Salihefendic