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

import os
import logging
import bleach
import tornado.web
import tornado.escape

from app.classes.models.server_permissions import EnumPermissionsServer
from app.classes.shared.console import Console
from app.classes.shared.helpers import Helpers
from app.classes.shared.file_helpers import FileHelpers
from app.classes.web.base_handler import BaseHandler

logger = logging.getLogger(__name__)


class FileHandler(BaseHandler):
    def render_page(self, template, page_data):
        self.render(
            template,
            data=page_data,
            translate=self.translator.translate,
        )

    @tornado.web.authenticated
    def get(self, page):
        api_key, _, exec_user = self.current_user
        superuser = exec_user["superuser"]
        if api_key is not None:
            superuser = superuser and api_key.superuser

        server_id = self.get_argument("id", None)

        permissions = {
            "Commands": EnumPermissionsServer.COMMANDS,
            "Terminal": EnumPermissionsServer.TERMINAL,
            "Logs": EnumPermissionsServer.LOGS,
            "Schedule": EnumPermissionsServer.SCHEDULE,
            "Backup": EnumPermissionsServer.BACKUP,
            "Files": EnumPermissionsServer.FILES,
            "Config": EnumPermissionsServer.CONFIG,
            "Players": EnumPermissionsServer.PLAYERS,
        }
        user_perms = self.controller.server_perms.get_user_id_permissions_list(
            exec_user["user_id"], server_id
        )

        if page == "get_file":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            file_path = Helpers.get_os_understandable_path(
                self.get_argument("file_path", None)
            )

            if not self.check_server_id(server_id, "get_file"):
                return
            else:
                server_id = bleach.clean(server_id)

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                file_path,
            ) or not Helpers.check_file_exists(os.path.abspath(file_path)):
                logger.warning(
                    f"Invalid path in get_file file file ajax call ({file_path})"
                )
                Console.warning(
                    f"Invalid path in get_file file file ajax call ({file_path})"
                )
                return

            error = None

            try:
                with open(file_path, encoding="utf-8") as file:
                    file_contents = file.read()
            except UnicodeDecodeError:
                file_contents = ""
                error = "UnicodeDecodeError"

            self.write({"content": file_contents, "error": error})
            self.finish()

        elif page == "get_tree":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            path = self.get_argument("path", None)

            if not self.check_server_id(server_id, "get_tree"):
                return
            else:
                server_id = bleach.clean(server_id)

            if Helpers.validate_traversal(
                self.controller.servers.get_server_data_by_id(server_id)["path"], path
            ):
                self.write(
                    Helpers.get_os_understandable_path(path)
                    + "\n"
                    + Helpers.generate_tree(path)
                )
            self.finish()

        elif page == "get_dir":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            path = self.get_argument("path", None)

            if not self.check_server_id(server_id, "get_tree"):
                return
            else:
                server_id = bleach.clean(server_id)

            if Helpers.validate_traversal(
                self.controller.servers.get_server_data_by_id(server_id)["path"], path
            ):
                self.write(
                    Helpers.get_os_understandable_path(path)
                    + "\n"
                    + Helpers.generate_dir(path)
                )
            self.finish()

    @tornado.web.authenticated
    def post(self, page):
        api_key, _, exec_user = self.current_user
        superuser = exec_user["superuser"]
        if api_key is not None:
            superuser = superuser and api_key.superuser

        server_id = self.get_argument("id", None)

        permissions = {
            "Commands": EnumPermissionsServer.COMMANDS,
            "Terminal": EnumPermissionsServer.TERMINAL,
            "Logs": EnumPermissionsServer.LOGS,
            "Schedule": EnumPermissionsServer.SCHEDULE,
            "Backup": EnumPermissionsServer.BACKUP,
            "Files": EnumPermissionsServer.FILES,
            "Config": EnumPermissionsServer.CONFIG,
            "Players": EnumPermissionsServer.PLAYERS,
        }
        user_perms = self.controller.server_perms.get_user_id_permissions_list(
            exec_user["user_id"], server_id
        )

        if page == "create_file":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            file_parent = Helpers.get_os_understandable_path(
                self.get_body_argument("file_parent", default=None, strip=True)
            )
            file_name = self.get_body_argument("file_name", default=None, strip=True)
            file_path = os.path.join(file_parent, file_name)

            if not self.check_server_id(server_id, "create_file"):
                return
            else:
                server_id = bleach.clean(server_id)

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                file_path,
            ) or Helpers.check_file_exists(os.path.abspath(file_path)):
                logger.warning(
                    f"Invalid path in create_file file ajax call ({file_path})"
                )
                Console.warning(
                    f"Invalid path in create_file file ajax call ({file_path})"
                )
                return

            # Create the file by opening it
            with open(file_path, "w", encoding="utf-8") as file_object:
                file_object.close()

        elif page == "create_dir":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            dir_parent = Helpers.get_os_understandable_path(
                self.get_body_argument("dir_parent", default=None, strip=True)
            )
            dir_name = self.get_body_argument("dir_name", default=None, strip=True)
            dir_path = os.path.join(dir_parent, dir_name)

            if not self.check_server_id(server_id, "create_dir"):
                return
            else:
                server_id = bleach.clean(server_id)

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                dir_path,
            ) or Helpers.check_path_exists(os.path.abspath(dir_path)):
                logger.warning(
                    f"Invalid path in create_dir file ajax call ({dir_path})"
                )
                Console.warning(
                    f"Invalid path in create_dir file ajax call ({dir_path})"
                )
                return
            # Create the directory
            os.mkdir(dir_path)

        elif page == "unzip_file":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            path = Helpers.get_os_understandable_path(self.get_argument("path", None))
            if Helpers.is_os_windows():
                path = Helpers.wtol_path(path)
            Helpers.unzip_file(path)
            self.redirect(f"/panel/server_detail?id={server_id}&subpage=files")
            return

    @tornado.web.authenticated
    def delete(self, page):
        api_key, _, exec_user = self.current_user
        superuser = exec_user["superuser"]
        if api_key is not None:
            superuser = superuser and api_key.superuser

        server_id = self.get_argument("id", None)

        permissions = {
            "Commands": EnumPermissionsServer.COMMANDS,
            "Terminal": EnumPermissionsServer.TERMINAL,
            "Logs": EnumPermissionsServer.LOGS,
            "Schedule": EnumPermissionsServer.SCHEDULE,
            "Backup": EnumPermissionsServer.BACKUP,
            "Files": EnumPermissionsServer.FILES,
            "Config": EnumPermissionsServer.CONFIG,
            "Players": EnumPermissionsServer.PLAYERS,
        }
        user_perms = self.controller.server_perms.get_user_id_permissions_list(
            exec_user["user_id"], server_id
        )
        if page == "del_file":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            file_path = Helpers.get_os_understandable_path(
                self.get_body_argument("file_path", default=None, strip=True)
            )

            Console.warning(f"Delete {file_path} for server {server_id}")

            if not self.check_server_id(server_id, "del_file"):
                return
            else:
                server_id = bleach.clean(server_id)

            server_info = self.controller.servers.get_server_data_by_id(server_id)
            if not (
                Helpers.in_path(
                    Helpers.get_os_understandable_path(server_info["path"]), file_path
                )
                or Helpers.in_path(
                    Helpers.get_os_understandable_path(server_info["backup_path"]),
                    file_path,
                )
            ) or not Helpers.check_file_exists(os.path.abspath(file_path)):
                logger.warning(f"Invalid path in del_file file ajax call ({file_path})")
                Console.warning(
                    f"Invalid path in del_file file ajax call ({file_path})"
                )
                return

            # Delete the file
            FileHelpers.del_file(file_path)

        elif page == "del_dir":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            dir_path = Helpers.get_os_understandable_path(
                self.get_body_argument("dir_path", default=None, strip=True)
            )

            Console.warning(f"Delete {dir_path} for server {server_id}")

            if not self.check_server_id(server_id, "del_dir"):
                return
            else:
                server_id = bleach.clean(server_id)

            server_info = self.controller.servers.get_server_data_by_id(server_id)
            if not Helpers.in_path(
                Helpers.get_os_understandable_path(server_info["path"]), dir_path
            ) or not Helpers.check_path_exists(os.path.abspath(dir_path)):
                logger.warning(f"Invalid path in del_file file ajax call ({dir_path})")
                Console.warning(f"Invalid path in del_file file ajax call ({dir_path})")
                return

            # Delete the directory
            # os.rmdir(dir_path)     # Would only remove empty directories
            if Helpers.validate_traversal(
                Helpers.get_os_understandable_path(server_info["path"]), dir_path
            ):
                # Removes also when there are contents
                FileHelpers.del_dirs(dir_path)

    @tornado.web.authenticated
    def put(self, page):
        api_key, _, exec_user = self.current_user
        superuser = exec_user["superuser"]
        if api_key is not None:
            superuser = superuser and api_key.superuser

        server_id = self.get_argument("id", None)
        permissions = {
            "Commands": EnumPermissionsServer.COMMANDS,
            "Terminal": EnumPermissionsServer.TERMINAL,
            "Logs": EnumPermissionsServer.LOGS,
            "Schedule": EnumPermissionsServer.SCHEDULE,
            "Backup": EnumPermissionsServer.BACKUP,
            "Files": EnumPermissionsServer.FILES,
            "Config": EnumPermissionsServer.CONFIG,
            "Players": EnumPermissionsServer.PLAYERS,
        }
        user_perms = self.controller.server_perms.get_user_id_permissions_list(
            exec_user["user_id"], server_id
        )
        if page == "save_file":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            file_contents = self.get_body_argument(
                "file_contents", default=None, strip=True
            )
            file_path = Helpers.get_os_understandable_path(
                self.get_body_argument("file_path", default=None, strip=True)
            )

            if not self.check_server_id(server_id, "save_file"):
                return
            else:
                server_id = bleach.clean(server_id)

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                file_path,
            ) or not Helpers.check_file_exists(os.path.abspath(file_path)):
                logger.warning(
                    f"Invalid path in save_file file ajax call ({file_path})"
                )
                Console.warning(
                    f"Invalid path in save_file file ajax call ({file_path})"
                )
                return

            # Open the file in write mode and store the content in file_object
            with open(file_path, "w", encoding="utf-8") as file_object:
                file_object.write(file_contents)

        elif page == "rename_file":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            item_path = Helpers.get_os_understandable_path(
                self.get_body_argument("item_path", default=None, strip=True)
            )
            new_item_name = self.get_body_argument(
                "new_item_name", default=None, strip=True
            )

            if not self.check_server_id(server_id, "rename_file"):
                return
            else:
                server_id = bleach.clean(server_id)

            if item_path is None or new_item_name is None:
                logger.warning("Invalid path(s) in rename_file file ajax call")
                Console.warning("Invalid path(s) in rename_file file ajax call")
                return

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                item_path,
            ) or not Helpers.check_path_exists(os.path.abspath(item_path)):
                logger.warning(
                    f"Invalid old name path in rename_file file ajax call ({server_id})"
                )
                Console.warning(
                    f"Invalid old name path in rename_file file ajax call ({server_id})"
                )
                return

            new_item_path = os.path.join(os.path.split(item_path)[0], new_item_name)

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                new_item_path,
            ) or Helpers.check_path_exists(os.path.abspath(new_item_path)):
                logger.warning(
                    f"Invalid new name path in rename_file file ajax call ({server_id})"
                )
                Console.warning(
                    f"Invalid new name path in rename_file file ajax call ({server_id})"
                )
                return

            # RENAME
            os.rename(item_path, new_item_path)

    @tornado.web.authenticated
    def patch(self, page):
        api_key, _, exec_user = self.current_user
        superuser = exec_user["superuser"]
        if api_key is not None:
            superuser = superuser and api_key.superuser

        server_id = self.get_argument("id", None)
        permissions = {
            "Commands": EnumPermissionsServer.COMMANDS,
            "Terminal": EnumPermissionsServer.TERMINAL,
            "Logs": EnumPermissionsServer.LOGS,
            "Schedule": EnumPermissionsServer.SCHEDULE,
            "Backup": EnumPermissionsServer.BACKUP,
            "Files": EnumPermissionsServer.FILES,
            "Config": EnumPermissionsServer.CONFIG,
            "Players": EnumPermissionsServer.PLAYERS,
        }
        user_perms = self.controller.server_perms.get_user_id_permissions_list(
            exec_user["user_id"], server_id
        )
        if page == "rename_file":
            if not permissions["Files"] in user_perms:
                if not superuser:
                    self.redirect("/panel/error?error=Unauthorized access to Files")
                    return
            item_path = Helpers.get_os_understandable_path(
                self.get_body_argument("item_path", default=None, strip=True)
            )
            new_item_name = self.get_body_argument(
                "new_item_name", default=None, strip=True
            )

            if not self.check_server_id(server_id, "rename_file"):
                return
            else:
                server_id = bleach.clean(server_id)

            if item_path is None or new_item_name is None:
                logger.warning("Invalid path(s) in rename_file file ajax call")
                Console.warning("Invalid path(s) in rename_file file ajax call")
                return

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                item_path,
            ) or not Helpers.check_path_exists(os.path.abspath(item_path)):
                logger.warning(
                    f"Invalid old name path in rename_file file ajax call ({server_id})"
                )
                Console.warning(
                    f"Invalid old name path in rename_file file ajax call ({server_id})"
                )
                return

            new_item_path = os.path.join(os.path.split(item_path)[0], new_item_name)

            if not Helpers.in_path(
                Helpers.get_os_understandable_path(
                    self.controller.servers.get_server_data_by_id(server_id)["path"]
                ),
                new_item_path,
            ) or Helpers.check_path_exists(os.path.abspath(new_item_path)):
                logger.warning(
                    f"Invalid new name path in rename_file file ajax call ({server_id})"
                )
                Console.warning(
                    f"Invalid new name path in rename_file file ajax call ({server_id})"
                )
                return

            # RENAME
            os.rename(item_path, new_item_path)

    def check_server_id(self, server_id, page_name):
        if server_id is None:
            logger.warning(
                f"Server ID not defined in {page_name} file ajax call ({server_id})"
            )
            Console.warning(
                f"Server ID not defined in {page_name} file ajax call ({server_id})"
            )
            return
        else:
            server_id = bleach.clean(server_id)

            # does this server id exist?
            if not self.controller.servers.server_id_exists(server_id):
                logger.warning(
                    f"Server ID not found in {page_name} file ajax call ({server_id})"
                )
                Console.warning(
                    f"Server ID not found in {page_name} file ajax call ({server_id})"
                )
                return
        return True