pusher_django_strategy.py

from django.http import HttpResponseForbidden
from django.http import JsonResponse

from pusher import Pusher


class PusherAuthStrategy:
    def __init__(self, namespace):
        self.namespace = namespace

    def authenticate(self, request, channel_name, socket_id, **kwargs):
        pusher_client = Pusher(APP_ID, API_KEY, SECRET_KEY, CLUSTER)
        if not request.user.is_authenticated:
            return HttpResponseForbidden()

        is_authorized = self._authenticate(
            request, channel_name, socket_id, **kwargs)

        if not is_authorized:
            return HttpResponseForbidden()

        kwargs = {
            'channel': channel_name,
            'socket_id': socket_id
        }
        if channel_name.startswith('presence'):
            kwargs['custom_data'] = {
                'user_id': request.user.id,
                'user_info': {
                    'username': request.user.username,
                    'first_name': request.user.first_name,
                    'last_name': request.user.last_name
                }
            }

        auth = pusher_client.authenticate(**kwargs)
        return JsonResponse(auth)

    def _authenticate(self, request, channel_name, socket_id, **kwargs):
        """Specific authentication per strategy"""
        raise NotImplementedError()


class NotAllowedStrategy(PusherAuthStrategy):
    def _authenticate(self, request, channel_name, socket_id):
        return False


class UserInTeamStrategy(PusherAuthStrategy):
    def _authenticate(self, request, channel_name, socket_id):
        team_name = channel_name.split(self.namespace)
        return request.user.is_member_of_team(team_name)


class UserPrivateStrategy(PusherAuthStrategy):
    def _authenticate(self, request, channel_name, socket_id):
        user_id = channel_name.split(self.namespace)
        return str(request.user.id) == user_id


AUTH_STRATEGIES = {
    'presence-team-': UserInTeamStrategy,
    'private-user-': UserPrivateStrategy
}
DEFAULT_STRATEGY = NotAllowedStrategy


def resolve_strategy(channel_name):
    for strategy_name, StrategyClass in AUTH_STRATEGIES.items():
        if channel_name.startswith(strategy_name):
            return StrategyClass(strategy_name)
    return DEFAULT_STRATEGY('')


def pusher_auth(request):
    channel_name = request.POST['channel_name']
    socket_id = request.POST['socket_id']
    strategy = resolve_strategy(channel_name)
    return strategy.authenticate(request, channel_name, socket_id)