mirror of
https://gitlab.com/crafty-controller/crafty-4.git
synced 2024-08-30 18:23:09 +00:00
4cc617fd0d
Tidy imports & dep auto-installer See merge request crafty-controller/crafty-commander!199
73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
import logging
|
|
from typing import (
|
|
Union,
|
|
List,
|
|
Optional, Tuple, Dict, Any
|
|
)
|
|
|
|
from app.classes.models.users import ApiKeys
|
|
from app.classes.shared.authentication import authentication
|
|
from app.classes.shared.main_controller import Controller
|
|
from app.classes.shared.helpers import helper
|
|
|
|
try:
|
|
import tornado.web
|
|
import bleach
|
|
|
|
except ModuleNotFoundError as e:
|
|
helper.auto_installer_fix(e)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class BaseHandler(tornado.web.RequestHandler):
|
|
|
|
nobleach = {bool, type(None)}
|
|
redactables = ("pass", "api")
|
|
|
|
# noinspection PyAttributeOutsideInit
|
|
def initialize(self, controller: Controller = None, tasks_manager=None, translator=None):
|
|
self.controller = controller
|
|
self.tasks_manager = tasks_manager
|
|
self.translator = translator
|
|
|
|
def get_remote_ip(self):
|
|
remote_ip = self.request.headers.get("X-Real-IP") or \
|
|
self.request.headers.get("X-Forwarded-For") or \
|
|
self.request.remote_ip
|
|
return remote_ip
|
|
|
|
current_user: Optional[Tuple[Optional[ApiKeys], Dict[str, Any], Dict[str, Any]]]
|
|
def get_current_user(self) -> Optional[Tuple[Optional[ApiKeys], Dict[str, Any], Dict[str, Any]]]:
|
|
return authentication.check(self.get_cookie("token"))
|
|
|
|
def autobleach(self, name, text):
|
|
for r in self.redactables:
|
|
if r in name:
|
|
logger.debug(f"Auto-bleaching {name}: [**REDACTED**]")
|
|
break
|
|
else:
|
|
logger.debug(f"Auto-bleaching {name}: {text}")
|
|
if type(text) in self.nobleach:
|
|
logger.debug("Auto-bleaching - bypass type")
|
|
return text
|
|
else:
|
|
return bleach.clean(text)
|
|
|
|
def get_argument(
|
|
self,
|
|
name: str,
|
|
default: Union[None, str, tornado.web._ArgDefaultMarker] = tornado.web._ARG_DEFAULT,
|
|
strip: bool = True,
|
|
) -> Optional[str]:
|
|
arg = self._get_argument(name, default, self.request.arguments, strip)
|
|
return self.autobleach(name, arg)
|
|
|
|
def get_arguments(self, name: str, strip: bool = True) -> List[str]:
|
|
if not isinstance(strip, bool):
|
|
raise AssertionError
|
|
args = self._get_arguments(name, self.request.arguments, strip)
|
|
args_ret = []
|
|
for arg in args:
|
|
args_ret += self.autobleach(name, arg)
|
|
return args_ret
|