skillbox/server/core/middleware.py

111 lines
3.8 KiB
Python

import json
import re
from django.conf import settings
from django.http import Http404, HttpResponsePermanentRedirect, HttpResponse
from django.shortcuts import redirect
from django.utils.deprecation import MiddlewareMixin
from core.utils import is_private_api_call_allowed
try:
from threading import local
except ImportError:
from django.utils._threading_local import local
_thread_locals = local()
def get_current_request():
""" returns the request object for this thread """
return getattr(_thread_locals, "request", None)
def get_current_user():
""" returns the current user, if exist, otherwise returns None """
request = get_current_request()
if request:
return getattr(request, "user", None)
class ThreadLocalMiddleware(MiddlewareMixin):
""" Simple middleware that adds the request object in thread local storage."""
def process_request(self, request):
_thread_locals.request = request
def process_response(self, request, response):
if hasattr(_thread_locals, 'request'):
del _thread_locals.request
return response
class CommonRedirectMiddleware(MiddlewareMixin):
"""
redirects common bad requests or missing images
"""
DEFAULT_REDIRECTS = [
(re.compile("(?i)(.+\.php|.+wp-admin.+|.+\.htm|\/wordpress\/|\/wp\/|\/mm5\/|\/wp-content\/)"),
'http://example.org/'),
]
def process_exception(self, request, exception):
if exception.__class__ == Http404:
new_path = self.not_found(request)
if new_path:
return redirect(new_path, permanent=True)
def process_response(self, request, response):
if response.status_code == 404:
new_path = self.not_found(request)
if new_path:
return HttpResponsePermanentRedirect(new_path)
return response
def not_found(self, request):
path = request.get_full_path().lower()
# generic stuff
for re_pattern, to in self.DEFAULT_REDIRECTS:
if re_pattern.match(path):
return to
if settings.USE_404_FALLBACK_IMAGE:
# some static image is missing show a placeholder (use full for local dev)
m = re.match(r".*(max|fill)-(?P<dimensions>\d+x\d+)\.(jpg|png|svg)", path)
if m:
return 'https://picsum.photos/{}'.format(m.group('dimensions').replace('x', '/'))
# or dummy image: return 'http://via.placeholder.com/{}'.format(m.group('dimensions'))
if '.png' in path or '.jpg' in path or '.svg' in path or 'not-found' in path:
return 'https://picsum.photos/400/400'
# https://stackoverflow.com/questions/4898408/how-to-set-a-login-cookie-in-django
class UserLoggedInCookieMiddleWare(MiddlewareMixin):
"""
Middleware to set user cookie
If user is authenticated and there is no cookie, set the cookie,
If the user is not authenticated and the cookie remains, delete it
"""
cookie_name = 'loginStatus'
def process_response(self, request, response):
#if user and no cookie, set cookie
if request.user.is_authenticated and not request.COOKIES.get(self.cookie_name):
response.set_cookie(self.cookie_name, 'true')
elif not request.user.is_authenticated and request.COOKIES.get(self.cookie_name):
#else if if no user and cookie remove user cookie, logout
response.delete_cookie(self.cookie_name)
return response
class UserHasLicenseMiddleWare(MiddlewareMixin):
def process_response(self, request, response):
if request.path == '/api/graphql/':
if not is_private_api_call_allowed(request.user, request.body):
return HttpResponse(json.dumps({'errors': ['no active license']}), status=402)
return response