Search code examples
pythondjangowebsocketdjango-channels

Django Channels permissions


Is there any supported way in Django Channels framework to write custom permission to connect to specific consumers?

something like DRF:

class MyConsumer(generics.APIView):
    permission_classes = [IsAuthenticated, MyCustomPermission]
    authentication_classes = [TokenAuthentication]

Solution

  • For anyone looking for a solution, I've come up with this solution:

    1. Create a Permissions class with a check_permissions method.
        from channels.exceptions import DenyConnection
    
        class Permissions(object):
            permission_classes = []
        
            def check_permissions(self):
                try:
                    for permission in self.permission_classes:
                        if permission(scope=self.scope).validate() != None:
                            raise DenyConnection
                except PermissionError:
                    raise DenyConnection
    
    1. Create a BasePermission class.
    class BasePermission(object):
        def __init__(self, scope, *args, **kwargs) -> None:
            self.scope = scope
    
        def has_permission(self, *args, **kwargs) -> bool:
            return True
    
        def validate(self, *args, **kwargs):
            try:
                if not self.has_permission(*args, **kwargs):
                    return PermissionError
            except Exception:
                return RuntimeError
            return None
    
    1. Then override has_perm with your custom permission.
    class NotAnonymousUser(BasePermission):
    
        def has_permission(self, *args, **kwargs) -> bool:
            try:
                return not isinstance(self.scope["user"],AnonymousUser)
            except Exception:
                return False
    
    1. Finally inherit the Permissions class and call the check_permissions method.
    class MyConsumer(Permissions, JsonWebsocketConsumer):
        permission_classes = [NotAnonymousUser]
    
        def connect(self):
            self.check_permissions()
                       .
                       .