mirror of
https://gitlab.com/bramw/baserow.git
synced 2025-03-12 11:46:54 +00:00
65 lines
2 KiB
Python
65 lines
2 KiB
Python
import uuid
|
|
from urllib.parse import parse_qs
|
|
|
|
from django.conf import settings
|
|
|
|
from channels.db import database_sync_to_async
|
|
from channels.middleware import BaseMiddleware
|
|
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
|
|
|
|
# Nosec disables spurious hardcoded password warning, this is not a password but instead
|
|
# the value of the JWT token to be used when a user wants to connect anonymously.
|
|
ANONYMOUS_USER_TOKEN = "anonymous" # nosec
|
|
|
|
|
|
@database_sync_to_async
|
|
def get_user(token):
|
|
"""
|
|
Selects a user related to the provided JWT token. If the token is invalid or if the
|
|
user does not exist then None is returned.
|
|
|
|
:param token: The JWT token for which the user must be fetched.
|
|
:type token: str
|
|
:return: The user related to the JWT token.
|
|
:rtype: User or None
|
|
"""
|
|
|
|
from baserow.api.user.jwt import get_user_from_token
|
|
|
|
anonymous = token == ANONYMOUS_USER_TOKEN
|
|
if anonymous:
|
|
if settings.DISABLE_ANONYMOUS_PUBLIC_VIEW_WS_CONNECTIONS:
|
|
return
|
|
else:
|
|
from django.contrib.auth.models import AnonymousUser
|
|
|
|
return AnonymousUser()
|
|
else:
|
|
try:
|
|
return get_user_from_token(token)
|
|
except (TokenError, InvalidToken):
|
|
return
|
|
|
|
|
|
class JWTTokenAuthMiddleware(BaseMiddleware):
|
|
"""
|
|
The auth middleware adds a user object to the scope if a valid JWT token is
|
|
provided via the GET parameters when requesting the web socket. It also adds a
|
|
unique web socket id for future identification.
|
|
"""
|
|
|
|
def __init__(self, inner):
|
|
self.inner = inner
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
get = parse_qs(scope["query_string"].decode("utf8"))
|
|
scope["user"] = None
|
|
scope["web_socket_id"] = None
|
|
|
|
jwt_token = get.get("jwt_token")
|
|
|
|
if jwt_token:
|
|
scope["user"] = await get_user(jwt_token[0])
|
|
scope["web_socket_id"] = str(uuid.uuid4())
|
|
|
|
return await self.inner(scope, receive, send)
|