Skip to content

Instantly share code, notes, and snippets.

@jamesgathu
Created May 8, 2019 08:07
Show Gist options
  • Save jamesgathu/049a01cd78b83385aadbc77b2ca1fa20 to your computer and use it in GitHub Desktop.
Save jamesgathu/049a01cd78b83385aadbc77b2ca1fa20 to your computer and use it in GitHub Desktop.
Incase you are using `djangorestframework-simplejwt` for authentication,
class JwtTokenAuthMiddleware(BaseMiddleware):
"""
JWT token authorization middleware for Django Channels 2
"""
def get_validated_token(self, raw_token):
"""
Validates an encoded JSON web token and returns a validated token
wrapper object.
"""
messages = []
for AuthToken in api_settings.AUTH_TOKEN_CLASSES:
try:
return AuthToken(raw_token)
except TokenError as e:
messages.append({'token_class': AuthToken.__name__,
'token_type': AuthToken.token_type,
'message': e.args[0]})
raise InvalidToken({
'detail': _('Given token not valid for any token type'),
'messages': messages,
})
def get_user(self, validated_token):
"""
Attempts to find and return a user using the given validated token.
"""
try:
user_id = validated_token[api_settings.USER_ID_CLAIM]
except KeyError:
raise InvalidToken(_('Token contained no recognizable user identification'))
try:
user = User.objects.get(**{api_settings.USER_ID_FIELD: user_id})
except User.DoesNotExist:
raise AuthenticationFailed(_('User not found'), code='user_not_found')
if not user.is_active:
raise AuthenticationFailed(_('User is inactive'), code='user_inactive')
return user
def __init__(self, inner):
self.inner = inner
def __call__(self, scope):
try:
raw_token = scope['query_string'].decode().split('=')[1]
validated_token = self.get_validated_token(raw_token)
user = self.get_user(validated_token=validated_token)
scope['user'] = user
except:
pass
return self.inner(scope)
JwtTokenAuthMiddlewareStack = lambda inner: JwtTokenAuthMiddleware(AuthMiddlewareStack(inner))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment