django-rest-framework/djangorestframework/permissions.py

"""
The :mod:`permissions` module bundles a set of  permission classes that are used 
for checking if a request passes a certain set of constraints. You can assign a permission 
class to your view by setting your View's :attr:`permissions` class attribute.
"""

from django.core.cache import cache
from djangorestframework import status
from djangorestframework.response import ErrorResponse
import time

__all__ = (
    'BasePermission',
    'FullAnonAccess',
    'IsAuthenticated',
    'IsAdminUser',
    'IsUserOrIsAnonReadOnly',
    'PerUserThrottling',
    'PerViewThrottling',
    'PerResourceThrottling'
)


_403_FORBIDDEN_RESPONSE = ErrorResponse(
    status.HTTP_403_FORBIDDEN,
    {'detail': 'You do not have permission to access this resource. ' +
               'You may need to login or otherwise authenticate the request.'})

_503_SERVICE_UNAVAILABLE = ErrorResponse(
    status.HTTP_503_SERVICE_UNAVAILABLE,
    {'detail': 'request was throttled'})


class ConfigurationException(BaseException):
    """To alert for bad configuration decisions as a convenience."""
    pass


class BasePermission(object):
    """
    A base class from which all permission classes should inherit.
    """
    def __init__(self, view):
        """
        Permission classes are always passed the current view on creation.
        """
        self.view = view
    
    def check_permission(self, auth):
        """
        Should simply return, or raise an :exc:`response.ErrorResponse`.
        """
        pass


class FullAnonAccess(BasePermission):
    """
    Allows full access.
    """

    def check_permission(self, user):
        pass


class IsAuthenticated(BasePermission):
    """
    Allows access only to authenticated users.
    """

    def check_permission(self, user):
        if not user.is_authenticated():
            raise _403_FORBIDDEN_RESPONSE 


class IsAdminUser(BasePermission):
    """
    Allows access only to admin users.
    """

    def check_permission(self, user):
        if not user.is_admin():
            raise _403_FORBIDDEN_RESPONSE


class IsUserOrIsAnonReadOnly(BasePermission):
    """
    The request is authenticated as a user, or is a read-only request.
    """

    def check_permission(self, user): 
        if (not user.is_authenticated() and
            self.view.method != 'GET' and
            self.view.method != 'HEAD'):
            raise _403_FORBIDDEN_RESPONSE


class BaseThrottle(BasePermission):
    """
    Rate throttling of requests.

    The rate (requests / seconds) is set by a :attr:`throttle` attribute
    on the :class:`.View` class.  The attribute is a string of the form 'number of
    requests/period'.

    Period should be one of: ('s', 'sec', 'm', 'min', 'h', 'hour', 'd', 'day')

    Previous request information used for throttling is stored in the cache.
    """    

    attr_name = 'throttle'
    default = '0/sec'
    timer = time.time

    def get_cache_key(self):
        """
        Should return a unique cache-key which can be used for throttling.
        Muse be overridden. 
        """
        pass

    def check_permission(self, auth):
        """
        Check the throttling.
        Return `None` or raise an :exc:`.ErrorResponse`.
        """
        num, period = getattr(self.view, self.attr_name, self.default).split('/')
        self.num_requests = int(num)
        self.duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
        self.auth = auth
        self.check_throttle()
        
    def check_throttle(self):
        """
        Implement the check to see if the request should be throttled.

        On success calls :meth:`throttle_success`.
        On failure calls :meth:`throttle_failure`.
        """
        self.key = self.get_cache_key()
        self.history = cache.get(self.key, [])
        self.now = self.timer()
        
        # Drop any requests from the history which have now passed the
        # throttle duration
        while self.history and self.history[0] <= self.now - self.duration:
            self.history.pop()

        if len(self.history) >= self.num_requests:
            self.throttle_failure()
        else:
            self.throttle_success()
    
    def throttle_success(self):
        """
        Inserts the current request's timestamp along with the key
        into the cache.
        """
        self.history.insert(0, self.now)
        cache.set(self.key, self.history, self.duration)
    
    def throttle_failure(self):
        """
        Called when a request to the API has failed due to throttling.
        Raises a '503 service unavailable' response.
        """
        raise _503_SERVICE_UNAVAILABLE


class PerUserThrottling(BaseThrottle):
    """
    Limits the rate of API calls that may be made by a given user.

    The user id will be used as a unique identifier if the user is
    authenticated. For anonymous requests, the IP address of the client will
    be used.
    """

    def get_cache_key(self):
        if self.auth.is_authenticated():
            ident = str(self.auth)
        else:
            ident = self.view.request.META.get('REMOTE_ADDR', None)
        return 'throttle_user_%s' % ident


class PerViewThrottling(BaseThrottle):
    """
    Limits the rate of API calls that may be used on a given view.

    The class name of the view is used as a unique identifier to
    throttle against.
    """

    def get_cache_key(self):
        return 'throttle_view_%s' % self.view.__class__.__name__

  
class PerResourceThrottling(BaseThrottle):
    """
    Limits the rate of API calls that may be used against all views on
    a given resource.

    The class name of the resource is used as a unique identifier to
    throttle against.
    """

    def get_cache_key(self):
        return 'throttle_resource_%s' % self.view.resource.__class__.__name__