codekingpro/portable-devtools

1##########################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##########################################################################9 10"""Implements File Manager"""11 12import os13import os.path14import secrets15import string16import time17from urllib.parse import unquote18from sys import platform as _platform19from flask_security import current_user20from pgadmin.utils.constants import ACCESS_DENIED_MESSAGE21import config22import codecs23import pathlib24 25import json26from flask import render_template, Response, session, request as req, \27    url_for, current_app, send_from_directory28from flask_babel import gettext29from pgadmin.user_login_check import pga_login_required30from pgadmin.utils import PgAdminModule31from pgadmin.utils import get_storage_directory32from pgadmin.utils.ajax import make_json_response, unauthorized, \33    internal_server_error34from pgadmin.utils.preferences import Preferences35from pgadmin.utils.constants import PREF_LABEL_OPTIONS, MIMETYPE_APP_JS, \36    MY_STORAGE37from pgadmin.settings.utils import get_file_type_setting38 39# Checks if platform is Windows40if _platform == "win32":41    import ctypes42    oldmode = ctypes.c_uint()43    kernel32 = ctypes.WinDLL('kernel32')44    SEM_FAILCRITICALERRORS = 145    SEM_NOOPENFILEERRORBOX = 0x800046    SEM_FAIL = SEM_NOOPENFILEERRORBOX | SEM_FAILCRITICALERRORS47    file_root = ""48 49MODULE_NAME = 'file_manager'50global transid51 52path_exists = os.path.exists53split_path = os.path.split54encode_json = json.JSONEncoder().encode55 56 57# utility functions58# convert bytes type to human readable format59def sizeof_fmt(num, suffix='B'):60    for unit in ['', 'k', 'M', 'G', 'T', 'P', 'E', 'Z']:61        if abs(num) < 1024.0:62            return "%3.1f %s%s" % (num, unit, suffix)63        num /= 1024.064    return "%.1f %s%s" % (num, 'Y', suffix)65 66 67# return size of file68def getsize(path):69    st = os.stat(path)70    return st.st_size71 72 73def getdrivesize(path):74    if _platform == "win32":75        free_bytes = ctypes.c_ulonglong(0)76        ctypes.windll.kernel32.GetDiskFreeSpaceExW(77            ctypes.c_wchar_p(path), None, None, ctypes.pointer(free_bytes))78        return free_bytes.value79 80 81# split extension for files82def splitext(path):83    for ext in ['.tar.gz', '.tar.bz2']:84        if path.endswith(ext):85            path, ext = path[:-len(ext)], path[-len(ext):]86            break87    else:88        path, ext = os.path.splitext(path)89    return ext[1:]90 91 92# check if file is hidden in windows platform93def is_folder_hidden(filepath):94    if _platform == "win32":95        try:96            attrs = ctypes.windll.kernel32.GetFileAttributesW(filepath)97            assert attrs != -198            result = bool(attrs & 2)99        except (AttributeError, AssertionError):100            result = False101        return result102    else:103        return os.path.basename(filepath).startswith('.')104 105 106class FileManagerModule(PgAdminModule):107    """108    FileManager lists files and folders and does109    following operations:110    - File selection111    - Folder selection112    - Open file113    - Create File114    and also supports:115    - Rename file116    - Delete file117    - Upload file118    - Create folder119    """120 121    LABEL = gettext("Storage")122 123    def get_own_menuitems(self):124        return {125            'file_items': []126        }127 128    def get_exposed_url_endpoints(self):129        """130        Returns:131            list: a list of url endpoints exposed to the client.132        """133        return [134            'file_manager.init',135            'file_manager.filemanager',136            'file_manager.index',137            'file_manager.delete_trans_id',138            'file_manager.save_last_dir',139            'file_manager.save_file_dialog_view',140            'file_manager.save_show_hidden_file_option'141        ]142 143    def get_file_size_preference(self):144        return self.file_upload_size145 146    def register_preferences(self):147        # Register 'file upload size' preference148        self.file_upload_size = self.preference.register(149            'options', 'file_upload_size',150            gettext("Maximum file upload size (MB)"), 'integer', 50,151            category_label=PREF_LABEL_OPTIONS152        )153        self.last_directory_visited = self.preference.register(154            'options', 'last_directory_visited',155            gettext("Last directory visited"), 'text', '/',156            category_label=PREF_LABEL_OPTIONS157        )158        self.last_storage = self.preference.register(159            'options', 'last_storage',160            gettext("Last storage"), 'text', '',161            category_label=PREF_LABEL_OPTIONS,162            hidden=True163        )164        self.file_dialog_view = self.preference.register(165            'options', 'file_dialog_view',166            gettext("File dialog view"), 'select', 'list',167            category_label=PREF_LABEL_OPTIONS,168            options=[{'label': gettext('List'), 'value': 'list'},169                     {'label': gettext('Grid'), 'value': 'grid'}],170            control_props={171                'allowClear': False,172                'tags': False173            },174        )175        self.show_hidden_files = self.preference.register(176            'options', 'show_hidden_files',177            gettext("Show hidden files and folders?"), 'boolean', False,178            category_label=PREF_LABEL_OPTIONS179        )180 181 182# Initialise the module183blueprint = FileManagerModule(MODULE_NAME, __name__)184 185 186@blueprint.route("/", endpoint='index')187@pga_login_required188def index():189    return bad_request(190        errormsg=gettext("This URL cannot be called directly.")191    )192 193 194@blueprint.route("/utility.js")195@pga_login_required196def utility():197    """render the required javascript"""198    return Response(response=render_template(199        "file_manager/js/utility.js", _=gettext),200        status=200,201        mimetype=MIMETYPE_APP_JS)202 203 204@blueprint.route(205    "/init", methods=["POST"], endpoint='init'206)207@pga_login_required208def init_filemanager():209    if len(req.data) != 0:210        configs = json.loads(req.data)211        trans_id = Filemanager.create_new_transaction(configs)212        data = Filemanager.get_trasaction_selection(trans_id)213        pref = Preferences.module('file_manager')214        file_dialog_view = pref.preference('file_dialog_view').get()215        if isinstance(file_dialog_view, list):216            file_dialog_view = file_dialog_view[0]217 218        last_selected_format = get_file_type_setting(data['supported_types'])219        # in some cases, the setting may not match with available types220        if last_selected_format not in data['supported_types']:221            last_selected_format = data['supported_types'][0]222 223        res_data = {224            'transId': trans_id,225            "options": {226                "culture": "en",227                "lang": "py",228                "defaultViewMode": file_dialog_view,229                "autoload": True,230                "showFullPath": False,231                "dialog_type": data['dialog_type'],232                "show_hidden_files":233                    pref.preference('show_hidden_files').get(),234                "fileRoot": data['fileroot'],235                "capabilities": data['capabilities'],236                "allowed_file_types": data['supported_types'],237                "platform_type": data['platform_type'],238                "show_volumes": data['show_volumes'],239                "homedir": data['homedir'],240                'storage_folder': data['storage_folder'],241                "last_selected_format": last_selected_format242            },243            "security": {244                "uploadPolicy": data['security']['uploadPolicy'],245                "uploadRestrictions": data['security']['uploadRestrictions'],246            },247            "upload": {248                "multiple": data['upload']['multiple'],249                "number": 20,250                "fileSizeLimit": data['upload']['fileSizeLimit'],251                "imagesOnly": False252            }253        }254 255    return make_json_response(data=res_data)256 257 258@blueprint.route(259    "/delete_trans_id/<int:trans_id>",260    methods=["DELETE"], endpoint='delete_trans_id'261)262@pga_login_required263def delete_trans_id(trans_id):264    Filemanager.release_transaction(trans_id)265    return make_json_response(266        data={'status': True}267    )268 269 270@blueprint.route(271    "/save_last_dir/<int:trans_id>", methods=["POST"], endpoint='save_last_dir'272)273@pga_login_required274def save_last_directory_visited(trans_id):275    blueprint.last_directory_visited.set(req.json['path'])276    blueprint.last_storage.set(req.json['storage_folder'])277    return make_json_response(status=200)278 279 280@blueprint.route(281    "/save_file_dialog_view/<int:trans_id>", methods=["POST"],282    endpoint='save_file_dialog_view'283)284@pga_login_required285def save_file_dialog_view(trans_id):286    blueprint.file_dialog_view.set(req.json['view'])287    return make_json_response(status=200)288 289 290@blueprint.route(291    "/save_show_hidden_file_option/<int:trans_id>", methods=["PUT"],292    endpoint='save_show_hidden_file_option'293)294@pga_login_required295def save_show_hidden_file_option(trans_id):296    blueprint.show_hidden_files.set(req.json['show_hidden'])297    return make_json_response(status=200)298 299 300class Filemanager():301    """FileManager Class."""302 303    # Stores list of dict for filename & its encoding304    loaded_file_encoding_list = []305 306    ERROR_NOT_ALLOWED = {307        'Error': gettext('Not allowed'),308        'Code': 0309    }310 311    def __init__(self, trans_id, ss=''):312        self.trans_id = trans_id313        self.dir = get_storage_directory()314        self.shared_dir = get_storage_directory(shared_storage=ss)315 316        if self.dir is not None and isinstance(self.dir, list):317            self.dir = ""318 319    @staticmethod320    def get_closest_parent(storage_dir, last_dir):321        """322        Check if path exists and if not then get closest parent which exists323        :param storage_dir: Base dir324        :param last_dir: check dir325        :return: exist dir326        """327        if len(last_dir) > 1 and \328                (last_dir.endswith('/') or last_dir.endswith('\\')):329            last_dir = last_dir[:-1]330        while last_dir:331            if os.path.exists(storage_dir or '' + last_dir):332                break333            index = max(last_dir.rfind('\\'), last_dir.rfind('/')) \334                if _platform == 'win32' else last_dir.rfind('/')335            last_dir = last_dir[0:index]336 337        if _platform == 'win32':338            if not last_dir.endswith('\\'):339                last_dir += "\\"340 341            return last_dir342 343        if not last_dir.endswith('/'):344            last_dir += "/"345 346        return last_dir347 348    @staticmethod349    def create_new_transaction(params):350        """351        It will also create a unique transaction id and352        store the information into session variable.353        Args:354            capabilities: Allow/Disallow user to perform355            selection, rename, delete etc.356        """357 358        # Define configs for dialog types359        # select file, select folder, create mode360        Filemanager.suspend_windows_warning()361        fm_type = params['dialog_type']362        storage_dir = get_storage_directory()363 364        # It is used in utitlity js to decide to365        # show or hide select file type options366        show_volumes = isinstance(storage_dir, list) or not storage_dir367        supp_types = allow_upload_files = params.get('supported_types', [])368 369        # tuples with (capabilities, files_only, folders_only, title)370        capability_map = {371            'select_file': (372                ['select_file', 'rename', 'upload', 'delete'],373                True,374                False,375                gettext("Select File")376            ),377            'select_folder': (378                ['select_folder', 'rename', 'create'],379                False,380                True,381                gettext("Select Folder")382            ),383            'create_file': (384                ['select_file', 'rename', 'create'],385                True,386                False,387                gettext("Create File")388            ),389            'storage_dialog': (390                ['select_folder', 'select_file', 'download',391                 'rename', 'delete', 'upload', 'create'],392                True,393                False,394                gettext("Storage Manager")395            ),396        }397 398        capabilities, files_only, folders_only, title = capability_map[fm_type]399 400        # Using os.path.join to make sure we have trailing '/' or '\'401        homedir = '/' if (config.SERVER_MODE) \402            else os.path.join(os.path.expanduser('~'), '')403 404        # get last visited directory, if not present then traverse in reverse405        # order to find closest parent directory406        if 'init_path' in params:407            blueprint.last_directory_visited.get(params['init_path'])408        last_dir = blueprint.last_directory_visited.get()409        last_ss_name = blueprint.last_storage.get()410        if last_ss_name and last_ss_name != MY_STORAGE \411                and len(config.SHARED_STORAGE) > 0:412            selected_dir = [sdir for sdir in config.SHARED_STORAGE if413                            sdir['name'] == last_ss_name]414            last_ss = selected_dir[0]['path'] if len(415                selected_dir) == 1 else storage_dir416        else:417            if last_ss_name != MY_STORAGE:418                last_dir = '/'419                blueprint.last_storage.set(MY_STORAGE)420 421            last_ss = storage_dir422 423        check_dir_exists = False424        if last_dir is None:425            last_dir = "/"426        else:427            check_dir_exists = True428 429        if not config.SERVER_MODE and last_dir == "/" or last_dir == "/":430            last_dir = homedir431 432        if check_dir_exists:433            last_dir = Filemanager.get_closest_parent(last_ss, last_dir)434 435        # create configs using above configs436        configs = {437            "fileroot": last_dir,438            "homedir": homedir,439            'storage_folder': last_ss_name,440            "dialog_type": fm_type,441            "title": title,442            "upload": {443                "multiple": True444            },445            "capabilities": capabilities,446            "security": {447                "uploadPolicy": "",448                "uploadRestrictions": allow_upload_files449            },450            "files_only": files_only,451            "folders_only": folders_only,452            "supported_types": supp_types,453            "platform_type": _platform,454            "show_volumes": show_volumes455        }456 457        # Create a unique id for the transaction458        trans_id = str(secrets.choice(range(1, 9999999)))459 460        if 'fileManagerData' not in session:461            file_manager_data = dict()462        else:463            file_manager_data = session['fileManagerData']464 465        file_upload_size = blueprint.get_file_size_preference().get()466        configs['upload']['fileSizeLimit'] = file_upload_size467        file_manager_data[trans_id] = configs468        session['fileManagerData'] = file_manager_data469        Filemanager.resume_windows_warning()470 471        return trans_id472 473    @staticmethod474    def get_trasaction_selection(trans_id):475        """476        This method returns the information of unique transaction477        id from the session variable.478 479        Args:480            trans_id: unique transaction id481        """482        file_manager_data = session['fileManagerData']483 484        # Return from the function if transaction id not found485        if str(trans_id) in file_manager_data:486            return file_manager_data[str(trans_id)]487 488    @staticmethod489    def release_transaction(trans_id):490        """491        This method is to remove the information of unique transaction492        id from the session variable.493 494        Args:495            trans_id: unique transaction id496        """497        file_manager_data = session['fileManagerData']498        # Return from the function if transaction id not found499        if str(trans_id) not in file_manager_data:500            return make_json_response(status=200)501 502        # Remove the information of unique transaction id503        # from the session variable.504        file_manager_data.pop(str(trans_id), None)505        session['fileManagerData'] = file_manager_data506 507        return make_json_response(status=200)508 509    @staticmethod510    def _get_drives_with_size(drive_name=None):511        """512        This is a generic function which returns the default path for storage513        manager dialog irrespective of any Platform type to list all514        files and directories.515        Platform windows:516        if no path is given, it will list volumes, else list directory517        Platform unix:518        it returns path to root directory if no path is specified.519        """520        def _get_drive_size(path):521            try:522                drive_size = getdrivesize(path)523                return sizeof_fmt(drive_size)524            except Exception:525                return 0526 527        if _platform == "win32":528            try:529                drives = []530                bitmask = ctypes.windll.kernel32.GetLogicalDrives()531                for letter in string.ascii_uppercase:532                    if bitmask & 1:533                        drives.append((letter, _get_drive_size(letter)))534                    bitmask >>= 1535                if (drive_name != '' and drive_name is not None and536                        drive_name in drives):537                    letter = "{0}{1}".format(drive_name, ':')538                    return (letter, _get_drive_size(letter))539                else:540                    return drives  # return drives if no argument is passed541            except Exception:542                return [('C:', _get_drive_size('C:'))]543        else:544            return '/'545 546    @staticmethod547    def suspend_windows_warning():548        """549        Prevents 'there is no disk in drive' waning on windows550        """551        # StackOverflow Ref: https://goo.gl/9gYdef552        if _platform == "win32":553            kernel32.SetThreadErrorMode(SEM_FAIL, ctypes.byref(oldmode))554 555    @staticmethod556    def resume_windows_warning():557        """558        Resumes waning on windows559        """560        if _platform == "win32":561            # Resume windows error562            kernel32.SetThreadErrorMode(oldmode, ctypes.byref(oldmode))563 564    @staticmethod565    def _skip_file_extension(566            file_type, supported_types, folders_only, file_extension):567        """568        Used internally by get_files_in_path to check if569        the file extn to be skipped570        """571        return file_type is not None and file_type != "*" and (572            folders_only or len(supported_types) > 0 and573            file_extension not in supported_types or574            file_type != file_extension)575 576    @staticmethod577    def get_files_in_path(578        show_hidden_files, files_only, folders_only, supported_types,579            file_type, user_dir, orig_path):580        """581        Get list of files and dirs in the path582        :param show_hidden_files: boolean583        :param files_only: boolean584        :param folders_only: boolean585        :param supported_types: array of supported types586        :param file_type: file type587        :param user_dir: base user dir588        :param orig_path: path after user dir589        :return:590        """591        files = []592 593        for f in sorted(os.listdir(orig_path)):594            system_path = os.path.join(os.path.join(orig_path, f))595 596            # continue if file/folder is hidden (based on user preference)597            if not show_hidden_files and is_folder_hidden(system_path):598                continue599            try:600                user_path = os.path.join(os.path.join(user_dir, f))601                created = time.ctime(os.path.getctime(system_path))602                modified = time.ctime(os.path.getmtime(system_path))603                file_extension = str(splitext(system_path))604            except Exception:605                continue606 607            # set protected to 1 if no write or read permission608            protected = 0609            if (not os.access(system_path, os.R_OK) or610                    not os.access(system_path, os.W_OK)):611                protected = 1612 613            # list files only or folders only614            if os.path.isdir(system_path):615                if files_only == 'true':616                    continue617                file_extension = "dir"618            # filter files based on file_type619            elif Filemanager._skip_file_extension(620                    file_type, supported_types, folders_only, file_extension):621                continue622 623            # create a list of files and folders624            files.append({625                "Filename": f,626                "Path": user_path,627                "file_type": file_extension,628                "Protected": protected,629                "Properties": {630                    "Date Created": created,631                    "Date Modified": modified,632                    "Size": sizeof_fmt(getsize(system_path))633                }634            })635 636        return files637 638    @staticmethod639    def list_filesystem(in_dir, path, trans_data, file_type, show_hidden):640        """641        It lists all file and folders within the given642        directory.643        """644        Filemanager.suspend_windows_warning()645        is_show_hidden_files = show_hidden646 647        path = unquote(path)648 649        Filemanager.check_access_permission(in_dir, path)650        Filemanager.resume_windows_warning()651 652        files = []653        if (_platform == "win32" and (path == '/' or path == '\\'))\654                and in_dir is None:655            drives = Filemanager._get_drives_with_size()656            for drive, drive_size in drives:657                path = file_name = "{0}:".format(drive)658                files.append({659                    "Filename": file_name,660                    "Path": path,661                    "file_type": 'drive',662                    "Protected": 1 if drive_size == 0 else 0,663                    "Properties": {664                        "Date Created": "",665                        "Date Modified": "",666                        "Size": drive_size667                    }668                })669            Filemanager.resume_windows_warning()670            return files671 672        orig_path = Filemanager.get_abs_path(in_dir, path)673 674        if not path_exists(orig_path):675            Filemanager.resume_windows_warning()676            return make_json_response(677                status=404,678                errormsg=gettext("'{0}' file does not exist.").format(path))679 680        user_dir = path681        folders_only = trans_data.get('folders_only', '')682        files_only = trans_data.get('files_only', '')683        supported_types = trans_data.get('supported_types', [])684 685        orig_path = unquote(orig_path)686        try:687            files = Filemanager.get_files_in_path(688                is_show_hidden_files, files_only, folders_only,689                supported_types, file_type, user_dir, orig_path690            )691        except Exception as e:692            Filemanager.resume_windows_warning()693            err_msg = str(e)694            if (hasattr(e, 'strerror') and695                    e.strerror == gettext('Permission denied')):696                err_msg = str(e.strerror)697            return unauthorized(err_msg)698        Filemanager.resume_windows_warning()699        return files700 701    @staticmethod702    def check_access_permission(in_dir, path, skip_permission_check=False):703        if not config.SERVER_MODE or skip_permission_check:704            return705 706        in_dir = '' if in_dir is None else in_dir707        orig_path = Filemanager.get_abs_path(in_dir, path)708 709        # This translates path with relative path notations710        # like ./ and ../ to absolute path.711        orig_path = os.path.abspath(orig_path)712 713        if in_dir:714            if _platform == 'win32':715                if in_dir[-1] == '\\' or in_dir[-1] == '/':716                    in_dir = in_dir[:-1]717            else:718                if in_dir[-1] == '/':719                    in_dir = in_dir[:-1]720 721        # Do not allow user to access outside his storage dir722        # in server mode.723        try:724            pathlib.Path(orig_path).relative_to(in_dir)725        except ValueError:726            raise PermissionError(gettext("Access denied ({0})").format(path))727 728    @staticmethod729    def get_abs_path(in_dir, path):730 731        if (path.startswith('\\\\') and _platform == 'win32')\732                or config.SERVER_MODE is False or in_dir is None:733            return "{}".format(path)734 735        if path == '/' or path == '\\':736            if _platform == 'win32':737                if in_dir.endswith('\\') or in_dir.endswith('/'):738                    return "{}".format(in_dir)739                else:740                    return "{}{}".format(in_dir, '\\')741            else:742                if in_dir.endswith('/'):743                    return "{}".format(in_dir)744                else:745                    return "{}{}".format(in_dir, '/')746 747        if in_dir.endswith('/') or in_dir.endswith('\\'):748            if path.startswith('/') or path.startswith('\\'):749                return "{}{}".format(in_dir[:-1], path)750            else:751                return "{}/{}".format(in_dir, path)752        else:753            if path.startswith('/') or path.startswith('\\'):754                return "{}{}".format(in_dir, path)755            else:756                return "{}/{}".format(in_dir, path)757 758    def validate_request(self, capability):759        """760        It validates the capability with the capabilities761        stored in the session762        """763        trans_data = Filemanager.get_trasaction_selection(self.trans_id)764        return False if capability not in trans_data['capabilities'] else True765 766    def getfolder(self, path=None, file_type="", show_hidden=False):767        """768        Returns files and folders in give path769        """770        trans_data = Filemanager.get_trasaction_selection(self.trans_id)771        the_dir = None772        if config.SERVER_MODE:773            if self.shared_dir and len(config.SHARED_STORAGE) > 0:774                the_dir = self.shared_dir775            else:776                the_dir = self.dir777 778            if the_dir is not None and not the_dir.endswith('/'):779                the_dir += '/'780 781        filelist = self.list_filesystem(782            the_dir, path, trans_data, file_type, show_hidden)783        return filelist784 785    def check_access(self, ss):786        if self.shared_dir:787            selected_dir_list = [sdir for sdir in config.SHARED_STORAGE if788                                 sdir['name'] == ss]789            selected_dir = selected_dir_list[0] if len(790                selected_dir_list) == 1 else None791 792            if selected_dir and selected_dir['restricted_access'] and \793                    not current_user.has_role("Administrator"):794                raise PermissionError(ACCESS_DENIED_MESSAGE)795 796    def rename(self, old=None, new=None):797        """798        Rename file or folder799        """800        if not self.validate_request('rename'):801            return unauthorized(self.ERROR_NOT_ALLOWED['Error'])802 803        if self.shared_dir:804            the_dir = self.shared_dir805        else:806            the_dir = self.dir if self.dir is not None else ''807 808        Filemanager.check_access_permission(the_dir, old)809        Filemanager.check_access_permission(the_dir, new)810 811        # check if it's dir812        if old[-1] == '/':813            old = old[:-1]814 815        # extract filename816        oldname = split_path(old)[-1]817        path = old818        path = split_path(path)[0]  # extract path819 820        if path[-1] != '/':821            path += '/'822 823        newname = new824        newpath = path + newname825 826        # make system old path827        oldpath_sys = "{0}{1}".format(the_dir, old)828        newpath_sys = "{0}{1}".format(the_dir, newpath)829 830        try:831            os.rename(oldpath_sys, newpath_sys)832        except OSError as e:833            return internal_server_error("{0} {1}".format(834                gettext('There was an error renaming the file:'), e.strerror))835 836        return {837            'Old Path': old,838            'Old Name': oldname,839            'New Path': newpath,840            'New Name': newname,841        }842 843    def delete(self, path=None):844        """845        Delete file or folder846        """847        if not self.validate_request('delete'):848            return unauthorized(self.ERROR_NOT_ALLOWED['Error'])849        if self.shared_dir:850            the_dir = self.shared_dir851        else:852            the_dir = self.dir if self.dir is not None else ''853        orig_path = "{0}{1}".format(the_dir, path)854 855        Filemanager.check_access_permission(the_dir, path)856 857        try:858            if os.path.isdir(orig_path):859                os.rmdir(orig_path)860            else:861                os.remove(orig_path)862        except OSError as e:863            return internal_server_error("{0} {1}".format(864                gettext('There was an error deleting the file:'), e.strerror))865 866        return make_json_response(status=200)867 868    def add(self, req=None):869        """870        File upload functionality871        """872        if not self.validate_request('upload'):873            return unauthorized(self.ERROR_NOT_ALLOWED['Error'])874 875        if self.shared_dir:876            the_dir = self.shared_dir877        else:878            the_dir = self.dir if self.dir is not None else ''879 880        try:881            path = req.form.get('currentpath')882 883            file_obj = req.files['newfile']884            file_name = file_obj.filename885            orig_path = "{0}{1}".format(the_dir, path)886            new_name = "{0}{1}".format(orig_path, file_name)887 888            try:889                # Check if the new file is inside the users directory890                if config.SERVER_MODE:891                    pathlib.Path(892                        os.path.abspath(893                            os.path.join(the_dir, new_name)894                        )895                    ).relative_to(the_dir)896            except ValueError:897                return unauthorized(self.ERROR_NOT_ALLOWED['Error'])898 899            with open(new_name, 'wb') as f:900                while True:901                    # 4MB chunk (4 * 1024 * 1024 Bytes)902                    data = file_obj.read(4194304)903                    if not data:904                        break905                    f.write(data)906        except OSError as e:907            return internal_server_error("{0} {1}".format(908                gettext('There was an error adding the file:'), e.strerror))909 910        Filemanager.check_access_permission(the_dir, path)911 912        return {913            'Path': path,914            'Name': new_name,915        }916 917    def is_file_exist(self, path, name):918        """919        Checks whether given file exists or not920        """921        the_dir = self.dir if self.dir is not None else ''922        code = 1923 924        name = unquote(name)925        path = unquote(path)926 927        orig_path = "{0}{1}".format(the_dir, path)928        Filemanager.check_access_permission(929            the_dir, "{}{}".format(path, name))930 931        new_name = "{0}{1}".format(orig_path, name)932        if not os.path.exists(new_name):933            code = 0934 935        return {936            'Path': path,937            'Name': name,938            'Code': code,939        }940 941    @staticmethod942    def get_new_name(in_dir, path, name):943        """944        Utility to provide new name for folder if file945        with same name already exists946        """947        new_name = name948        count = 0949        while True:950            if not (path.endswith("/") or name.startswith("/")):951                path = path + "/"952            file_path = "{}{}/".format(path, new_name)953            create_path = file_path954            if in_dir != "":955                create_path = "{}/{}".format(in_dir, file_path)956 957            if not path_exists(create_path):958                return create_path, file_path, new_name959            else:960                count += 1961                new_name = "{}_{}".format(name, count)962 963    @staticmethod964    def check_file_for_bom_and_binary(filename, enc="utf-8"):965        """966        This utility function will check if file is Binary file967        and/or if it startswith BOM character968 969        Args:970            filename: File971            enc: Encoding for the file972 973        Returns:974            Status(Error?), Error message, Binary file flag,975            BOM character flag and Encoding to open file976        """977        status = True978        err_msg = None979        is_startswith_bom = False980        is_binary = False981 982        # check if file type is text or binary983        text_chars = bytearray([7, 8, 9, 10, 12, 13, 27]) \984            + bytearray(range(0x20, 0x7f)) \985            + bytearray(range(0x80, 0x100))986 987        def is_binary_string(bytes_data):988            """Checks if string data is binary"""989            return bool(990                bytes_data.translate(None, text_chars)991            )992 993        # read the file994        try:995 996            with open(filename, 'rb') as f:997                file_data = f.read(1024)998 999            # Check for BOM in file data1000            for encoding, boms in \1001                    ('utf-8-sig', (codecs.BOM_UTF8,)), \1002                    ('utf-16', (codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)), \1003                    ('utf-32', (codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE)):1004                if any(file_data.startswith(bom) for bom in boms):1005                    is_startswith_bom = True1006                    enc = encoding1007 1008            # No need to check for binary file, a BOM marker already1009            # indicates that text stream afterwards1010            if not is_startswith_bom:1011                # Check if string is binary1012                is_binary = is_binary_string(file_data)1013 1014            # Store encoding for future use1015            Filemanager.loaded_file_encoding_list.\1016                append({os.path.basename(filename): enc})1017 1018        except IOError as ex:1019            # we don't want to expose real path of file1020            # so only show error message.1021            if ex.strerror == 'Permission denied':1022                return unauthorized(str(ex.strerror))1023            else:1024                return internal_server_error(str(ex.strerror))1025 1026        except Exception as ex:1027            return internal_server_error(str(ex.strerror))1028 1029        # Remove root storage path from error message1030        # when running in Server mode1031        if not status and not current_app.PGADMIN_RUNTIME:1032            storage_directory = get_storage_directory()1033            if storage_directory:1034                err_msg = err_msg.replace(storage_directory, '')1035 1036        return status, err_msg, is_binary, is_startswith_bom, enc1037 1038    def addfolder(self, path, name):1039        """1040        Functionality to create new folder1041        """1042        if not self.validate_request('create'):1043            return unauthorized(self.ERROR_NOT_ALLOWED['Error'])1044 1045        if self.shared_dir and len(config.SHARED_STORAGE) > 0:1046            user_dir = self.shared_dir1047        else:1048            user_dir = self.dir if self.dir is not None else ''1049 1050        Filemanager.check_access_permission(user_dir, "{}{}".format(1051            path, name))1052 1053        create_path, new_path, new_name = \1054            self.get_new_name(user_dir, path, name)1055        try:1056            os.mkdir(create_path)1057        except OSError as e:1058            return internal_server_error(str(e.strerror))1059 1060        result = {1061            'Parent': path,1062            'Path': new_path,1063            'Name': new_name,1064            'Date Modified': time.ctime(time.time())1065        }1066 1067        return result1068 1069    def download(self, path=None):1070        """1071        Functionality to download file1072        """1073        if not self.validate_request('download'):1074            return unauthorized(self.ERROR_NOT_ALLOWED['Error'])1075 1076        if self.shared_dir and len(config.SHARED_STORAGE) > 0:1077            the_dir = self.shared_dir1078        else:1079            the_dir = self.dir if self.dir is not None else ''1080 1081        orig_path = "{0}{1}".format(the_dir, path)1082 1083        Filemanager.check_access_permission(1084            the_dir, "{}{}".format(path, path)1085        )1086 1087        filename = os.path.basename(path)1088        if orig_path and len(orig_path) > 0:1089            dir_path = os.path.dirname(orig_path)1090        else:1091            dir_path = os.path.dirname(path)1092 1093        response = send_from_directory(dir_path, filename,1094                                       mimetype='application/octet-stream',1095                                       as_attachment=True)1096        response.headers["filename"] = filename1097 1098        return response1099 1100    def permission(self, path=None):1101        the_dir = self.dir if self.dir is not None else ''1102        res = {'Code': 1}1103        Filemanager.check_access_permission(the_dir, path)1104        return res1105 1106 1107@blueprint.route(1108    "/filemanager/<int:trans_id>/",1109    methods=["POST"], endpoint='filemanager'1110)1111@pga_login_required1112def file_manager(trans_id):1113    """1114    It is the common function for every call which is made1115    and takes function name from post request and calls it.1116    It gets unique transaction id from post request and1117    rotate it into Filemanager class.1118    """1119    mode = ''1120    kwargs = {}1121    if req.method == 'POST':1122        if req.files:1123            mode = 'add'1124            kwargs = {'req': req,1125                      'storage_folder': req.form.get('storage_folder', None)}1126        else:1127            kwargs = json.loads(req.data)1128            kwargs['req'] = req1129            mode = kwargs['mode']1130            del kwargs['mode']1131    elif req.method == 'GET':1132        kwargs = {1133            'path': req.args['path'],1134            'name': req.args['name'] if 'name' in req.args else ''1135        }1136        mode = req.args['mode']1137    ss = kwargs['storage_folder'] if 'storage_folder' in kwargs else None1138    my_fm = Filemanager(trans_id, ss)1139 1140    if ss and mode in ['upload', 'rename', 'delete', 'addfolder', 'add',1141                       'permission']:1142        my_fm.check_access(ss)1143    func = getattr(my_fm, mode)1144    try:1145        if mode in ['getfolder', 'download']:1146            kwargs.pop('name', None)1147 1148        if mode in ['add']:1149            kwargs.pop('storage_folder', None)1150 1151        if mode in ['addfolder', 'getfolder', 'rename', 'delete',1152                    'is_file_exist', 'req', 'permission', 'download']:1153            kwargs.pop('req', None)1154            kwargs.pop('storage_folder', None)1155 1156        res = func(**kwargs)1157    except PermissionError as e:1158        return unauthorized(str(e))1159 1160    if isinstance(res, Response):1161        return res1162    return make_json_response(data={'result': res, 'status': True})1163