# Copyright (c) 2013, 2019, Oracle and/or its affiliates. All rights reserved. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License, version 2.0, # as published by the Free Software Foundation. # # This program is also distributed with certain software (including # but not limited to OpenSSL) that is licensed under separate terms, as # designated in a particular file or component or in included license # documentation. The authors of MySQL hereby grant you an additional # permission to link the program and your derivative works with the # separately licensed software that they have included with MySQL. # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See # the GNU General Public License, version 2.0, for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import os import sys import ctypes if sys.platform == 'win32': from ctypes import wintypes import errno import subprocess import shutil import stat from wb_common import PermissionDeniedError class FunctionType: Success = 0 Boolean = 1 String = 2 Data = 3 class FileUtils(object): """ Provides implementation of functions to perform operations on the Windows File System. Exception handling is expected to be done by the callers: - FileOpsLocalWindows class for operations done as the CURRENT user - The wbadminhelper for operations done as ADMIN Some exception handling is done on these functions but only to be able to generate the PermissionDeniedError as it is used on further validations on the admin code. The functions can be divided in the next groups: - Success functions: they are expected to succeed, they don't return any value - Boolean functions: on success execution return a boolean value - String functions: on success execution return a string The functions will return the proper value on success or generate an exception in case of failure. The function names are self descriptive, however the function type will be indicated on each of them using the criteria explained above """ def __init__(self): pass @classmethod def get_free_space(self, path): """ Function Type : String """ def get_readable_format(total): measures = ['B', 'KB', 'MB', 'GB', 'TB'] index = 0 while index < len(measures) and total > 1024: total = float(total / 1024) index = index + 1 return "%.2f %s" % (total, measures[index]) total_bytes, free_bytes = ctypes.c_ulonglong(0), ctypes.c_ulonglong(0) if ctypes.windll.kernel32.GetDiskFreeSpaceExW(ctypes.c_wchar_p(path), None, ctypes.pointer(total_bytes), ctypes.pointer(free_bytes)): free = get_readable_format(free_bytes.value) total = get_readable_format(total_bytes.value) ret_val = "%s of %s available" % (free, total) else: ret_val = "Could not determine" return ret_val @classmethod def check_dir_writable(self, path): """ Function Type : Boolean """ ret_val = True if os.path.isdir(path): dummy_file = os.path.join(path, '.wb_write_test') try: f = open(dummy_file, 'w+') f.close() os.remove(dummy_file) except IOError as e: if e.errno == errno.EACCES: ret_val = False else: raise elif os.path.exists(path): raise IOError(errno.ENOTDIR, 'The indicated path is not a directory') else: raise IOError(errno.ENOENT, 'The indicated path does not exist') return ret_val @classmethod def check_path_exists(self, path): """ Function Type : Boolean """ return os.path.exists(path) @classmethod def check_file_readable(self, path): """ Function Type : Boolean """ return os.access(path, os.R_OK) @classmethod def create_directory(self, path): """ Function Type : Success """ try: os.mkdir(path) except (IOError, OSError) as err: if err.errno == errno.EACCES: raise PermissionDeniedError("Could not create directory %s" % path) raise @classmethod def create_directory_recursive(self, path): """ Function Type : Success """ try: os.makedirs(path) except (IOError, OSError) as err: if err.errno == errno.EACCES: raise PermissionDeniedError("Could not create directory %s" % path) raise @classmethod def remove_directory(self, path): """ Function Type : Success """ try: os.rmdir(path) except (IOError, OSError) as err: if err.errno == errno.EACCES: raise PermissionDeniedError("Could not remove directory %s" % path) raise err @classmethod def remove_directory_recursive(self, path): """ Function Type : Success """ try: shutil.rmtree(path) except (IOError, OSError) as err: if err.errno == errno.EACCES: raise PermissionDeniedError("Could not remove directory %s" % path) raise err @classmethod def copy_file(self, source, target, target_backup = ""): """ Function Type : Success """ try: # Does a backup if it was required if target_backup: shutil.copy(target, target_backup) shutil.copy(source, target) except (IOError, OSError) as e: if e.errno == errno.EACCES: raise PermissionDeniedError("Can't copy %s to %s" % (source, target)) raise @classmethod def delete_file(self, path): """ Function Type : Success """ try: os.remove(path) except (IOError, OSError) as err: if err.errno == errno.EACCES: raise PermissionDeniedError("Could not delete file %s" % path) raise err @classmethod def get_file_owner(self, path): """ Function Type : String """ # Grabs the functions that will be used to locate the owner of a specific path GetFileSecurityW = ctypes.windll.advapi32.GetFileSecurityW GetSecurityDescriptorOwner = ctypes.windll.advapi32.GetSecurityDescriptorOwner LookupAccountSidW = ctypes.windll.advapi32.LookupAccountSidW lookup_account = False get_owner_descriptor = False ret_val = "" SECURITY_INFORMATION_OWNER = 0X00000001 length = wintypes.DWORD() GetFileSecurityW.restype = wintypes.BOOL GetFileSecurityW.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, ctypes.POINTER(wintypes.BYTE), wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ] # First: retrieves the security descriptor for the requested path # This call is to retrieve the size needed to store this descriptor GetFileSecurityW(path, SECURITY_INFORMATION_OWNER, None, 0, ctypes.byref(length)) if length.value: sd = (wintypes.BYTE * length.value)() # This call retrieves the security descriptor if GetFileSecurityW(path, SECURITY_INFORMATION_OWNER, sd, length, ctypes.byref(length)): get_owner_descriptor = True # Second: Now gets the owner information based on the security descriptor retrieved # on the first step if get_owner_descriptor: sdo = ctypes.POINTER(wintypes.BYTE)() sdo_defaulted = wintypes.BOOL() if GetSecurityDescriptorOwner(sd, ctypes.byref(sdo), ctypes.byref(sdo_defaulted)): lookup_account = True # Finally: using the security descriptor owner retrieved above, queries # the user naeme and the first domain where the user is found if lookup_account: SIZE = 256 name = ctypes.create_unicode_buffer(SIZE) domain = ctypes.create_unicode_buffer(SIZE) cch_name = wintypes.DWORD(SIZE) cch_domain = wintypes.DWORD(SIZE) sdo_type = wintypes.DWORD() if LookupAccountSidW(None, sdo, name, ctypes.byref(cch_name), domain, ctypes.byref(cch_domain), ctypes.byref(sdo_type)): ret_val = "%s\\%s" % (domain.value, name.value) else: print('Failed to lookup user') if not ret_val: raise IOError(errno.EINVAL, "The given path is not a file or directory") return ret_val @classmethod def list_dir(self, path, include_size, output_handler): try: # This is needed on the SudoTail class if os.path.isfile(path): if include_size: stat_info = os.stat(path) stat_info.st_size line = "%i %s" % (stat_info.st_size, path) output_handler(line) else: output_handler(path) else: dlist = os.listdir(path) for item in dlist: line = "" item_path = os.path.join(path, item) stat_info = os.stat(item_path) item_stat = stat_info.st_mode if stat.S_ISDIR(item_stat): item += '/' if include_size: line = "%s %s" % (str(stat_info.st_size), item) else: line = item elif stat.S_ISREG(item_stat) or stat.S_ISLNK(item_stat): if include_size: line = "%s %s" % (str(stat_info.st_size), item) else: line = item if line: output_handler(line) except (IOError, OSError) as e: if e.errno == errno.EACCES: raise PermissionDeniedError("Permission denied accessing %s" % path) raise @classmethod def get_file_lines(self, path, skip, output_handler): try: f = open(path, 'r') skipped = 0 for line in f: if not skip or skipped == skip: output_handler(line) else: skipped = skipped + 1 f.close() except (IOError, OSError) as e: if e.errno == errno.EACCES: raise PermissionDeniedError("Can't open file '%s'" % path) raise class OSUtils(object): """ Provides the exec_command functiono wich will execute whatever command is passed as long as it is valid for the OS. In case of failure this function will also raise the proper exception """ @classmethod def exec_command(self, command, output_handler): """ Executes any OS valid command and sends the output to the command listener Syntax: EXEC command: A valida OS command. """ retcode = 1 try: # Note that self._command contains the internal command so EXEC # and self._args is the real OS command to be executed # Executes the command, reading the output from the PIPE process = subprocess.Popen(command, stdin = subprocess.PIPE, encoding="utf-8", stdout = subprocess.PIPE, stderr = subprocess.STDOUT, shell=True) # Sends all the command output to the listener if output_handler: for line in iter(process.stdout.readline, ""): output_handler(line) # Waits for the process completion process.wait() retcode = process.returncode except Exception as exc: raise return retcode