import logging
import re
import typing as t
import orjson
import nh3
import tornado.web
from app.classes.models.crafty_permissions import EnumPermissionsCrafty
from app.classes.models.server_permissions import EnumPermissionsServer
from app.classes.models.users import ApiKeys
from app.classes.helpers.helpers import Helpers
from app.classes.helpers.file_helpers import FileHelpers
from app.classes.helpers.embed_helpers import build_embed_meta
from app.classes.shared.main_controller import Controller
from app.classes.shared.translation import Translation
from app.classes.shared.main_models import DatabaseShortcuts
from app.classes.models.users import DoesNotExist
logger = logging.getLogger(__name__)
auth_log = logging.getLogger("auth")
bearer_pattern = re.compile(r"^Bearer ", flags=re.IGNORECASE)
class BaseHandler(tornado.web.RequestHandler):
nobleach = {bool, type(None)}
redactables = ("pass", "api")
helper: Helpers
controller: Controller
translator: Translation
file_helper: FileHelpers
# noinspection PyAttributeOutsideInit
def initialize(
self,
helper=None,
controller=None,
tasks_manager=None,
translator=None,
file_helper=None,
):
self.helper = helper
self.controller = controller
self.tasks_manager = tasks_manager
self.translator = translator
self.file_helper = file_helper
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()
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
def get_embed_meta(self, page_path):
"""Build OG embed metadata for a public page, or None if disabled."""
settings = self.controller.management.get_embed_settings()
base_url = f"{self.request.protocol}://{self.request.host}"
return build_embed_meta(settings, base_url, page_path)
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.
"""
try:
return self.controller.authentication.check(self.get_cookie("token"))
except DoesNotExist:
return None
def autobleach(self, name, text):
for r in self.redactables:
if r in name:
logger.debug(f"Auto-bleaching {name}: [**REDACTED**]")
break
logger.debug(f"Auto-bleaching {name}: {text}")
if type(text) in self.nobleach:
logger.debug("Auto-bleaching - bypass type")
return text
return nh3.clean(text) # pylint: disable=no-member
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)
bleached = self.autobleach(name, arg)
if "&" in str(bleached):
bleached = bleached.replace("&", "&")
return bleached
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 is_totp_method(self):
if (
re.match(
r"^/api/v2/(users/.+/totp/|users/[^/]+/totp/.+/verify/)$",
self.request.path,
)
and self.request.method == "POST"
) or (
re.match(r"^/api/v2/users/[^/]+/totp/recovery/renew/$", self.request.path)
and self.request.method == "GET"
):
return True
return False
def is_mfa_not_present(self, user, token_data) -> bool:
"""Checks to see if panel settings or role settings require user
to have MFA enabled and passed in token. Checks token to see
if user has signed in with MFA.
Args:
user (dict): dictionary of user object
token_data (dict): decoded token data
Returns:
bool: Returns False if user has signed in with MFA or if they have not and
it is not required. Returns True if user is required to have MFA and
has not signed in with it.
"""
su_mfa = self.helper.get_setting(
"superMFA"
) # Get super user forced MFA setting
role_mfa = False
for role in self.controller.users.get_user_roles_id(user["user_id"]):
if self.controller.roles.get_role(role)["mfa_required"]:
role_mfa = True
break
return ((user["superuser"] and su_mfa) or role_mfa) and not token_data.get(
"mfa"
)
def authenticate_user(
self,
) -> t.Optional[
t.Tuple[
t.List,
t.List[EnumPermissionsCrafty],
t.List[str],
bool,
t.Dict[str, t.Any],
str,
]
]:
try:
api_key, token_data, user = self.controller.authentication.check_err(
self._auth_get_api_token()
)
if not user["enabled"]:
# Log the user out and send error if disabled
self.clear_cookie("token")
return self.finish_json(
403,
{
"status": "error",
"error": "ACCESS_DENIED",
"error_data": self.helper.translation.translate(
"login", "accountDisabled", user["lang"]
),
},
)
if token_data.get("token_id") and token_data.get(
"session_id"
): # Check to see if token defines session_id and token_id
return self.finish_json(
403,
{
"status": "error",
"error": "ACCESS_DENIED",
"error_data": self.helper.translation.translate(
"error", "duplicateId", user["lang"]
),
},
)
if self.is_mfa_not_present(user, token_data) and (
not self.is_totp_method()
and not token_data.get("token_id")
and user["username"] != "anti-lockout-user"
):
# check to see if user is superuser
# and MFA is not in token.
# Also check to see if user is trying to add MFA or access backup codes.
# Check for token ID because only API keys will have this.
warning = self.helper.translation.translate(
"otp", "mfaWarn", user["lang"]
)
goto = self.helper.translation.translate(
"otp", "goToPage", user["lang"]
)
url = f"/panel/edit_user_otp?id={user['user_id']}"
return self.finish_json(
403,
{
"status": "error",
"error": "ACCESS_DENIED",
"error_data": (f"{warning}
{goto}"),
},
)
superuser = user["superuser"]
server_permissions_api_mask = ""
if api_key is not None:
superuser = superuser and api_key.full_access
server_permissions_api_mask = api_key.server_permissions
if api_key.full_access:
server_permissions_api_mask = "1" * len(EnumPermissionsServer)
exec_user_role = set()
if superuser:
authorized_servers = self.controller.servers.get_all_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?
)
authorized_servers = [
DatabaseShortcuts.get_data_obj(x.server_object)
for x in authorized_servers
]
logger.debug("Checking results")
if user:
return (
authorized_servers,
exec_user_crafty_permissions,
exec_user_role,
superuser,
user,
server_permissions_api_mask,
)
logging.debug("Auth unsuccessful")
auth_log.error(
f"Authentication attempted from {self.get_remote_ip()}. Invalid token"
)
self.access_denied(None, "the user provided an invalid token")
return None
except Exception as auth_exception:
auth_log.error(
f"Authentication attempted from {self.get_remote_ip()}."
f" Error: {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",
"error_data": "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))