crafty-4/app/classes/web/base_handler.py

242 lines
7.9 KiB
Python

import logging
import re
import typing as t
import orjson
import bleach
import tornado.web
from app.classes.models.crafty_permissions import EnumPermissionsCrafty
from app.classes.models.users import ApiKeys
from app.classes.shared.helpers import Helpers
from app.classes.shared.main_controller import Controller
from app.classes.shared.translation import Translation
logger = logging.getLogger(__name__)
bearer_pattern = re.compile(r"^Bearer ", flags=re.IGNORECASE)
class BaseHandler(tornado.web.RequestHandler):
def set_default_headers(self) -> None:
"""
Fix CORS
"""
self.set_header("Access-Control-Allow-Origin", "*")
self.set_header(
"Access-Control-Allow-Headers",
"Content-Type, x-requested-with, Authorization",
)
self.set_header(
"Access-Control-Allow-Methods", "POST, GET, PUT, DELETE, OPTIONS"
)
def options(self, *_, **__):
"""
Fix CORS
"""
# no body
self.set_status(204)
self.finish()
nobleach = {bool, type(None)}
redactables = ("pass", "api")
helper: Helpers
controller: Controller
translator: Translation
# noinspection PyAttributeOutsideInit
def initialize(
self, helper=None, controller=None, tasks_manager=None, translator=None
):
self.helper = helper
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: t.Tuple[t.Optional[ApiKeys], t.Dict[str, t.Any], t.Dict[str, t.Any]]
"""
A variable that contains the current user's data. Please see
Please only use this with routes using the `@tornado.web.authenticated` decorator.
"""
def get_current_user(
self,
) -> t.Optional[
t.Tuple[t.Optional[ApiKeys], t.Dict[str, t.Any], t.Dict[str, t.Any]]
]:
"""
Get the token's API key, the token's payload and user data.
Returns:
t.Optional[ApiKeys]: The API key of the token.
t.Dict[str, t.Any]: The token's payload.
t.Dict[str, t.Any]: The user's data from the database.
"""
return self.controller.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: t.Union[
None, str, tornado.web._ArgDefaultMarker
] = tornado.web._ARG_DEFAULT,
strip: bool = True,
) -> t.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) -> t.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
def access_denied(self, user: t.Optional[str], reason: t.Optional[str]):
ip = self.get_remote_ip()
route = self.request.path
if user is not None:
user_data = f"User {user} from IP {ip}"
else:
user_data = f"An unknown user from IP {ip}"
if reason:
ending = f"to the API route {route} because {reason}"
else:
ending = f"to the API route {route}"
logger.info(f"{user_data} was denied access {ending}")
self.finish_json(
403,
{
"status": "error",
"error": "ACCESS_DENIED",
"info": "You were denied access to the requested resource",
},
)
def _auth_get_api_token(self) -> t.Optional[str]:
"""Get an API token from the request
The API token is searched in the following order:
1. The `token` query parameter
2. The `Authorization` header
3. The `token` cookie
Returns:
t.Optional[str]: The API token or None if no token was found.
"""
logger.debug("Searching for specified token")
api_token = self.get_query_argument("token", None)
if api_token is None and self.request.headers.get("Authorization"):
api_token = bearer_pattern.sub(
"", self.request.headers.get("Authorization")
)
elif api_token is None:
api_token = self.get_cookie("token")
return api_token
def authenticate_user(
self,
) -> t.Optional[
t.Tuple[
t.List,
t.List[EnumPermissionsCrafty],
t.List[str],
bool,
t.Dict[str, t.Any],
]
]:
try:
api_key, _token_data, user = self.controller.authentication.check_err(
self._auth_get_api_token()
)
superuser = user["superuser"]
if api_key is not None:
superuser = superuser and api_key.superuser
exec_user_role = set()
if superuser:
authorized_servers = self.controller.list_defined_servers()
exec_user_role.add("Super User")
exec_user_crafty_permissions = (
self.controller.crafty_perms.list_defined_crafty_permissions()
)
else:
if api_key is not None:
exec_user_crafty_permissions = (
self.controller.crafty_perms.get_api_key_permissions_list(
api_key
)
)
else:
exec_user_crafty_permissions = (
self.controller.crafty_perms.get_crafty_permissions_list(
user["user_id"]
)
)
logger.debug(user["roles"])
for r in user["roles"]:
role = self.controller.roles.get_role(r)
exec_user_role.add(role["role_name"])
authorized_servers = self.controller.servers.get_authorized_servers(
user["user_id"] # TODO: API key authorized servers?
)
logger.debug("Checking results")
if user:
return (
authorized_servers,
exec_user_crafty_permissions,
exec_user_role,
superuser,
user,
)
else:
logging.debug("Auth unsuccessful")
self.access_denied(None, "the user provided an invalid token")
return None
except Exception as auth_exception:
logger.debug(
"An error occured while authenticating an API user:",
exc_info=auth_exception,
)
self.finish_json(
403,
{
"status": "error",
"error": "ACCESS_DENIED",
"info": "An error occured while authenticating the user",
},
)
return None
def finish_json(self, status: int, data: t.Dict[str, t.Any]):
self.set_status(status)
self.set_header("Content-Type", "application/json")
self.finish(orjson.dumps(data)) # pylint: disable=no-member