File manager - Edit - /home/ferretapmx/public_html/python-packages.tar
Back
mysqlsh/util.py 0000644 00000002401 15231164450 0007571 0 ustar 00 # Copyright (c) 2021, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 def to_dict(obj): """Recursively convert a built-in Shell object to native types""" pass mysqlsh/plugin_manager/repositories.py 0000644 00000033153 15231164450 0014343 0 ustar 00 # Copyright (c) 2020, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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.path import json import mysqlsh from mysqlsh.plugin_manager import plugin, plugin_function from . import general # Define default repositories DEFAULT_PLUGIN_REPOSITORIES = [{ "name": "Official MySQL Shell Plugin Repository", "description": "The official MySQL Shell Plugin Repository maintained by " "the MySQL Team at Oracle.", "url": "https://cdn.mysql.com/mysql_shell/plugins_manifest.zip", "manifestPath": "manifest/mysql-shell-plugins-manifest.json" }] def format_repository_listing(repositories): """Returns a formatted list of repositories. Args: repositories (list): A list of repositories. Returns: The formated list as string """ out = "" i = 1 for r in repositories: out += (f"{i:>4} {r.get('name', '??')}\n" f" {r.get('description', '-')}\n" f" {r.get('url', '??')}\n\n") i += 1 return out def get_user_repositories(raise_exceptions=False): """Fetches the registered user repositories Args: raise_exceptions (bool): Whether exceptions are raised Returns: The registered repositories as list """ user_repo_file_path = os.path.join(general.get_shell_user_dir(), "plugin-repositories.json") if os.path.isfile(user_repo_file_path): try: with open(user_repo_file_path, 'r') as user_repo_file: user_repos = json.load(user_repo_file).get("repositories") return user_repos except json.JSONDecodeError as e: print( f"ERROR: Could not parse JSON file '{user_repo_file_path}'.\n" f"{str(e.msg)}, line: {e.lineno}, col: {e.colno}") if raise_exceptions: raise return except OSError as e: print(f"ERROR: Error reading file '{user_repo_file_path}'.\n" f"{str(e)}") if raise_exceptions: raise else: return [] def set_user_repositories(user_repos, raise_exceptions=False): """Stores the registered user repositories Args: user_repos (list): The list of raise_exceptions (bool): Whether exceptions are raised Returns: True on success """ user_repo_file_content = { "fileType": "MySQL Shell Plugin Repositories", "version": "0.0.1", "repositories": user_repos } user_repo_file_path = os.path.join(general.get_shell_user_dir(), "plugin-repositories.json") try: with open(user_repo_file_path, 'w') as user_repo_file: json.dump(user_repo_file_content, user_repo_file, indent=4) return True except OSError as e: print(f"ERROR: Error writing to file '{user_repo_file_path}'.\n" f"{str(e)}") if raise_exceptions: raise return False @plugin_function('plugins.repositories.list') def get_plugin_repositories(**kwargs): """Lists all registered plugin repositories This function will list all registered MySQL Shell plugin repositories. To add a repository use the plugins.repositories.add() function. Args: **kwargs: Optional parameters Keyword Args: return_formatted (bool): If set to true, a list object is returned. raise_exceptions (bool): Whether exceptions are raised Returns: None or a list of dicts representing the plugins """ return_formatted = kwargs.get('return_formatted', True) raise_exceptions = kwargs.get('raise_exceptions', False) # Init the repo list with the default repositories repos = DEFAULT_PLUGIN_REPOSITORIES.copy() user_repos = get_user_repositories(raise_exceptions=raise_exceptions) if user_repos: repos.extend(user_repos) if not return_formatted: return repos else: # cSpell:ignore repositor print("Registered MySQL Shell Plugin Repositories.\n\n" f"{format_repository_listing(repositories=repos)}" f"Total of {len(repos)} repositor" f"{'y' if len(repos) == 1 else 'ies'}.\n") @plugin_function('plugins.repositories.add') def add_plugin_repository(url=None, **kwargs): """Adds a new plugin repository By calling this function a new plugin repository can be added. The url parameter allows for shortcuts. 'domain.com' expands to https://domain.com/mysql-shell-plugins-manifest.json 'domain.com/plugins' expands to https://domain.com/plugins/mysql-shell-plugins-manifest.json 'github/username' expands to https://raw.githubusercontent.com/username/mysql-shell-plugins/master/mysql-shell-plugins-manifest.json 'github/username/repo' expands to https://raw.githubusercontent.com/username/repo/master/mysql-shell-plugins-manifest.json Args: url (str): The url of a MySQL Shell plugins repository. **kwargs: Optional parameters Keyword Args: interactive (bool): Whether user input is accepted raise_exceptions (bool): Whether exceptions are raised Returns: None """ interactive = kwargs.get('interactive', True) raise_exceptions = kwargs.get('raise_exceptions', True) if not url and interactive: print("To add a new MySQL Shell plugin repository the URL of the " "repository manifest file\nneeds to be specified.\n\n" "Examples:\n" " domain.com\n" " domain.com/plugins\n" " github/username\n" " github/username/repository\n") url = mysqlsh.globals.shell.prompt( f"Please enter the URL of the plugin repository: ", { 'defaultValue': '' }).strip() if not url: print("No URL given. Cancelling Operation") return if url.lower().startswith("github/"): github_str = url.split("/") if len(github_str) == 1: print("A github user needs to be specified") return github_user = github_str[1] # Get the github repo if len(github_str) == 2: github_repo = "mysql-shell-plugins" else: github_repo = github_str[2] url = ( f"https://raw.githubusercontent.com/{github_user}/{github_repo}/" f"master/mysql-shell-plugins-manifest.json") else: # Add https:// if not specified if not url.startswith("https://") and not url.startswith("http://"): url = f"https://{url}" if not url.endswith(".json"): if url.endswith("/"): url = f"{url}mysql-shell-plugins-manifest.json" else: url = f"{url}/mysql-shell-plugins-manifest.json" try: manifest = general.download_file(url) if not manifest: return manifest_content = json.loads(manifest) repository = manifest_content.get("repository") name = repository.get('name') if not name: print("The given repository misses the 'name' property.") return description = repository.get('description') if not name: print("The given repository misses the 'description' property.") return plugins = manifest_content.get('plugins') if not plugins or len(plugins) == 0: print("The repository does not contain any plugins. " "Operation cancelled.") return # Warn the user about possible consequences print("\nWARNING:\n" "You are about to add an external MySQL Shell plugin repository." "\nExternal plugin repositories and their plugins complement\n" "the functionality of MySQL Shell and can contain system\n" "level software that could be potentially harmful to your\n" "system. Please review the description below and only proceed\n" "if you have obtained the external plugin repository URL from\n" "a trusted source.\n\n" "Oracle and its affiliates cannot be held responsible for\n" "any potential harm caused by using plugins from external " "sources.\n") # Ensure the user will explicitly confirm the full URL print(f"{'Repository :'} {name}\n" f"{'Description:'} {description}\n" f"URL: {url}\n\n" "The repository contains the following plugins:\n") for plugin in plugins: print(f" - {plugin.get('caption')}\n") prompt = mysqlsh.globals.shell.prompt( f"Are you sure you want to add the repository '{name}'" " [yes/NO]: ", { 'defaultValue': 'no' }).strip().lower() if prompt != 'yes': print("Operation cancelled.") return print("Fetching current user repositories...") user_repos = get_user_repositories(raise_exceptions=raise_exceptions) if user_repos is None: return # Check if the given repository is already in the list for r in user_repos: if r.get("url").lower() == url.lower(): print(f"The repository '{name}' has already been added.\n") return print(f"Adding repository '{name}'...") user_repos.append({ "name": name, "description": description, "url": url }) if set_user_repositories(user_repos): print(f"Repository '{name}' successfully added.\n") except json.JSONDecodeError as e: print(f"ERROR: Could not parse JSON file. {str(e.msg)}, " f"line: {e.lineno}, col: {e.colno}") if raise_exceptions: raise return def search_user_repo(url, user_repos): # Search by name nr = 0 found = False for r in user_repos: if r.get("url") == url: found = True break nr += 1 if found: return nr else: return None @plugin_function('plugins.repositories.remove') def remove_plugin_repository(**kwargs): """Removes a registered plugin repository This function will remove a shell plugin repository previously registered. Args: **kwargs: Optional parameters Keyword Args: url (str): The url of a MySQL Shell plugins repository. interactive (bool): Whether user input is accepted raise_exceptions (bool): Whether exceptions are raised Returns: None """ url = kwargs.get('url') interactive = kwargs.get('interactive', True) raise_exceptions = kwargs.get('raise_exceptions', False) user_repos = get_user_repositories(raise_exceptions=raise_exceptions) if not user_repos: print("No custom MySQL Shell plugin repositories registered.\n") return repo_index = None if url: repo_index = search_user_repo(url, user_repos) if repo_index is None: print(f"No user repository matches '{url}'") if repo_index is None and interactive: print("Removing a plugin repository.\n\n" f"{format_repository_listing(user_repos)}") # Let the user choose from the list while repo_index is None: prompt = mysqlsh.globals.shell.prompt( "Please enter the index or URL of a repository: ").strip() if prompt: try: try: # If the user provided an index, try to map that nr = int(prompt) if nr > 0 and nr <= len(user_repos): repo_index = nr - 1 else: raise IndexError except ValueError: repo_index = search_user_repo(prompt, user_repos) if repo_index is None: raise ValueError except (ValueError, IndexError): print(f"The repository '{prompt}' was not found. Please try " "again or leave empty to cancel the operation.\n") else: break if repo_index is None: print("No valid URL provided. Cancelling operation.") else: url = user_repos[repo_index].get("url", '') print(f"\nRemoving repository '{url}'...") del user_repos[repo_index] if set_user_repositories(user_repos): print(f"Repository successfully removed.\n") mysqlsh/plugin_manager/plugins.py 0000644 00000073020 15231164450 0013272 0 ustar 00 # Copyright (c) 2020, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 json import os import re import shutil import zipfile import mysqlsh from enum import Enum from . import general from . import repositories from mysqlsh.plugin_manager import validate_shell_version class Filter(Enum): NONE = "all plugins" ONLY_INSTALLED = "installed plugins" NOT_INSTALLED = "available plugins for installation" UPDATABLE = "updatable plugins" def parse_version(version): ret_val = tuple(int(x) if x.isnumeric() else x for x in version.split(".")) return ret_val def get_repositories_with_plugins(plugin_filter=Filter.NONE, printouts=True, raise_exceptions=False): """Downloads all plugin information from the available repositories Args: printouts (bool): Whether to print out information raise_exceptions (bool): Whether exceptions should be raised Returns: A list of repositories including their plugins """ if printouts: print(f"Fetching list of {plugin_filter.value}...\n") # Get the list of all repositories, cSpell:ignore repos repos = repositories.get_plugin_repositories(return_formatted=False) try: i = 1 for r in repos: # If the manifest is stored in a zip, extact that zip and get the # manifest file if r.get("url").endswith(".zip"): # Download the zip file zip_file_path = general.get_shell_temp_dir("manifest.zip") # If there is an earlier manifest.zip, remove it if os.path.isfile(zip_file_path): os.remove(zip_file_path) if not general.download_file(r.get("url"), zip_file_path): continue # Extract the zip file manifest_dir = general.get_shell_user_dir( "temp", "manifest_zip") if os.path.isdir(manifest_dir): shutil.rmtree(manifest_dir) try: with zipfile.ZipFile(zip_file_path, 'r') as zip_ref: zip_ref.extractall(manifest_dir) except zipfile.BadZipFile as e: print(f"ERROR: Could not extract zip file {zip_file_path}." f"\n{str(e)}") if raise_exceptions: raise # Remove zip file os.remove(zip_file_path) zip_manifest_path = r.get("manifestPath") if not zip_manifest_path: zip_manifest_path = "mysql-shell-plugins-manifest.json" manifest_path = manifest_dir for path_item in zip_manifest_path.split('/'): manifest_path = os.path.join(manifest_path, path_item) if not os.path.isfile(manifest_path): shutil.rmtree(manifest_dir) print(f"ERROR: Cannot open file '{manifest_path}'.") return with open(manifest_path, "r") as manifest_file: manifest = manifest_file.read() shutil.rmtree(manifest_dir) else: manifest = general.download_file(r.get("url")) # Load the current plugin manifest of the repo if manifest: plugins = json.loads(manifest).get("plugins") # Add id, installed, installedVersion, # installedDevelopmentStage and updateAvailable fields to all # plugins for p in plugins: p["id"] = i i += 1 # Set installed, installedVersion and updateAvailable name = p.get("name") installed = name in dir(mysqlsh.globals) p["installed"] = installed if installed: latest_version = p.get("latestVersion") plugin = getattr(mysqlsh.globals, name) if plugin and "version" in dir(plugin): installed_version = plugin.version() p["installedVersion"] = installed_version if parse_version(latest_version) > \ parse_version(installed_version): p["updateAvailable"] = True version_data = get_plugin_version_data( p, installed_version) if version_data: installed_dev_stage = version_data.get( "developmentStage") if installed_dev_stage: p["installedDevelopmentStage"] = \ installed_dev_stage.upper() # Filter out installed plugins if needed if plugin_filter == Filter.ONLY_INSTALLED: plugins = [p for p in plugins if p.get("installed")] elif plugin_filter == Filter.NOT_INSTALLED: plugins = [p for p in plugins if not p.get("installed")] elif plugin_filter == Filter.UPDATABLE: plugins = [ p for p in plugins if p.get("updateAvailable", False) ] # Add list of plugins to the repo r["plugins"] = plugins return repos except json.JSONDecodeError as e: print(f"ERROR: Could not parse JSON file. {str(e.msg)}, " f"line: {e.lineno}, col: {e.colno}") if raise_exceptions: raise except Exception as e: print(f"ERROR: Could not get the list of repositories.\n{str(e)}") if raise_exceptions: raise def get_plugin_version_data(plugin, version): """Returns the plugin version data for a given version Args: plugin (dict): A dict representing a plugin version (str): The version to look up Returns: The version data as a dict or None if the version is not found """ versions = plugin.get("versions") if not versions: return for v in versions: if v.get("version") == version: return v def get_plugin_count(repositories): """Returns the total number of plugin Args: repositories (dict): A list of repositories with their plugins. Returns: The number of plugins """ i = 0 if repositories: for r in repositories: plugins = r.get("plugins") if not plugins: continue i += len(plugins) return i def get_installed_plugin_count(repositories): """Returns the total number of plugin Args: repositories (dict): A list of repositories with their plugins. Returns: The number of plugins """ i = 0 for r in repositories: plugins = r.get("plugins") if not plugins: continue for p in plugins: if p.get("installed", False): i += 1 return i def get_available_update_count(repositories): """Returns the number of available plugin updates Args: repositories (dict): A list of repositories with their plugins. print_info (bool): If set to True, information about the updates will be printed Returns: A tuple containing the number of available updates and info text """ updateable_count = 0 for r in repositories: plugins = r.get("plugins") if not plugins: continue for p in plugins: if p.get("updateAvailable", False): updateable_count += 1 if updateable_count == 1: out = (f"One update is available. " "Use plugins.update() to install the update.") elif updateable_count > 1: out = (f"{updateable_count} updates are available. " "Use plugins.update() to install the updates.") else: out = "No updates available." return updateable_count, out def format_plugin_listing(repositories, add_description=True): """Returns a formatted list of plugins. Args: repositories (list): A list of repositories with their plugins. Returns: The formated list as string """ i = 1 out = "" header = (f" # {'Name':20} {'Caption':34} {'Version':16} " f"{'Installed':16}") sep = f"---- {'-'*20} {'-'*34} {'-'*16} {'-'*16}" for r in repositories: plugins = r.get("plugins") if not plugins: continue repo_name = r.get('name', 'Repository: ?') if out == "": out += f"{header}\n{sep}\n" else: if not add_description: out += '\n' #out += f"{repo_name}\n{'='*94}\n" for p in plugins: id = p.get("id", i) name = p.get("name") name = name[:18] + '..' if len(name) > 20 else name # Caption caption = p.get("caption") caption = re.sub( r'[\n\r]', ' ', caption[:32] + '..' if len(caption) > 34 else caption) # Description desc = p.get("description") desc = re.sub(r'[\n\r]', ' ', desc[:73] + '..' if len(desc) > 75 else desc) # Latest Version latest_v = p.get("latestVersion") # Development Stage dev_stage = "" version_data = get_plugin_version_data(p, latest_v) if version_data: dev_stage = version_data.get("developmentStage") if dev_stage: dev_stage = dev_stage.upper() version_str = latest_v + " " + dev_stage # Installed Version installed = p.get("installed", False) installed_version = p.get("installedVersion", "?.?.?") installed_dev_stage = p.get("installedDevelopmentStage", "??") update_available = "^" if p.get("updateAvailable", False) else "" installed_str = installed_version + " " + installed_dev_stage + \ update_available if installed: out += f"*{id:>3} " else: out += f"{id:>4} " out += (f"{name:20} {caption:34} {version_str:16} " f"{installed_str if installed else 'No':16}\n") if add_description: out += f" {desc}\n\n" i += 1 return out def get_plugin(repositories, name=None, interactive=True): """Returns a plugin dict Args: repositories (list): A list of repositories with their plugins. name (str): The name of the plugin interactive (bool): Whether user input is accepted Returns: A dict representing the plugin """ selected_item = None plugin_count = get_plugin_count(repositories) # If there are no plugins, return None if plugin_count == 0: print("No plugins available.") return # If no name is given but there is only one plugin, return that one # if name is None and plugin_count == 1: # for r in repositories: # plugins = r.get("plugins") # if not plugins: # continue # if len(plugins) > 0: # selected_item = plugins[0] if name is None and interactive: out = format_plugin_listing(repositories, add_description=False) print(out) # Let the user choose from the list while selected_item is None: prompt = mysqlsh.globals.shell.prompt( "Please enter the index or name of a plugin: ").strip().lower( ) if not prompt: print("Operation cancelled.") return plugin_count = get_plugin_count(repositories) try: try: # If the user provided an index, try to map that nr = int(prompt) if nr > 0 and nr <= plugin_count: for r in repositories: plugins = r.get("plugins") if not plugins: continue for p in plugins: if nr == p.get("id"): selected_item = p break if selected_item: break else: raise IndexError except ValueError: # Search by name for r in repositories: plugins = r.get("plugins") if not plugins: continue for p in plugins: if prompt == p.get("name"): selected_item = p break if selected_item: break if selected_item is None: raise ValueError except (ValueError, IndexError): print(f"The plugin '{prompt}' was not found. Please try again " "or leave empty to cancel the operation.\n") return # Add empty line print("") else: for r in repositories: plugins = r.get("plugins") if not plugins: continue for p in plugins: if name == p.get("name"): selected_item = p break if selected_item: break if not selected_item: print(f"The plugin '{name}' was not found.") return selected_item def list_plugins(**kwargs): """Lists plugins all available MySQL Shell plugins. This function will list all all available plugins in the registered plugin repositories. To add a new plugin repository use the plugins.repositories.add() function. Args: **kwargs: Optional parameters Keyword Args: return_formatted (bool): If set to true, a list object is returned. interactive (bool): Whether user input is accepted Returns: None or a list of dicts representing the plugins """ return_formatted = kwargs.get('return_formatted', True) interactive = kwargs.get('interactive', True) # Get available repositories including the plugin lists repositories = get_repositories_with_plugins() if not repositories: return if not return_formatted: return repositories else: # Get number of available updates updates_available, update_info = get_available_update_count( repositories) out = format_plugin_listing(repositories) plugin_count = get_plugin_count(repositories=repositories) inst_count = get_installed_plugin_count(repositories=repositories) if inst_count > 0: out += (f"* {inst_count} plugin{'s' if inst_count > 1 else ''} " "installed, ") if plugin_count > 0: out += (f"{plugin_count} plugin" f"{'s' if plugin_count > 1 else ''}" " total.") # If there are available updates, add that information if updates_available > 0: out += f"\n^ {update_info}" if interactive: out += ( "\n\nUse plugins.details() to get more information about a " "specific plugin.\n") else: out += "\n" return out def install_plugin(name=None, **kwargs): """Installs a MySQL Shell plugin. This function download and install a plugin Args: name (str): The name of the plugin. **kwargs: Optional parameters Keyword Args: version (str): If specified, that specific version of the plugin will be installed force_install (bool): It set to true will first remove the plugin if it already exists return_object (bool): Whether to return the object interactive (bool): Whether user input is accepted printouts (bool): Whether information should be printed raise_exceptions (bool): Whether exceptions are raised Returns: None or plugin information """ version = kwargs.get("version") force_install = kwargs.get("force_install") return_object = kwargs.get("return_object", False) interactive = kwargs.get("interactive", True) printouts = kwargs.get("printouts", True) raise_exceptions = kwargs.get("raise_exceptions", False) try: # Get the list of repositories and their plugins repositories = get_repositories_with_plugins( plugin_filter=Filter.NONE if name or version or force_install else Filter.NOT_INSTALLED, printouts=printouts) if not repositories: return # Get the plugin by name or interactive user selection plugin = get_plugin(repositories, name, interactive) if plugin is None: return # Get plugin name and caption name = plugin.get('name') caption = plugin.get('caption') module_name = plugin.get('moduleName') installed = name in dir(mysqlsh.globals) if installed and not force_install: print(f"The plugin '{name}' is already installed. Use the " "force_install parameter to re-install it anyway.\n") return # Get version data version = version if version else plugin.get('latestVersion') version_data = get_plugin_version_data(plugin=plugin, version=version) if not version_data: print( f"ERROR: Version {version} not found for plugin {caption}.\n") return shell_version_min = version_data.get('shellVersionMin',None) shell_version_max = version_data.get('shellVersionMax',None) try: validate_shell_version(min=shell_version_min, max=shell_version_max) except Exception as e: print(str(e)) return if printouts: print(f"Installing {caption} ...") # Get the right download URL for the plugin urls = version_data.get('urls') if not urls: print(f"ERROR: No URLs available for plugin {caption}.\n") return # TODO: Check if the shell binary is a community or commercial build url = urls.get('community') zip_file_name = url.split('/')[-1] zip_file_path = general.get_shell_temp_dir(zip_file_name) # If the zip file already exists, delete it if os.path.isfile(zip_file_path): os.remove(zip_file_path) # Download the file if not general.download_file(url=url, file_path=zip_file_path): return # Set plugin directory using the module_name of the plugin plugin_dir = general.get_plugins_dir(module_name) # If the plugin folder already exists, delete it if force_install and os.path.isdir(plugin_dir): shutil.rmtree(plugin_dir) # Extract the zip file try: with zipfile.ZipFile(zip_file_path, 'r') as zip_ref: zip_ref.extractall(plugin_dir) except zipfile.BadZipFile as e: print( f"ERROR: Could not extract zip file {zip_file_path}.\n{str(e)}" ) if raise_exceptions: raise # Remove zip file os.remove(zip_file_path) if printouts: print(f"{caption} has been installed successfully.\n\n" f"Please restart the shell to load the plugin. To get help " f"type '\\? {name}' after restart.\n") if return_object: return plugin except Exception as e: print(f"ERROR: Could not install the plugin.\n{str(e)}") if raise_exceptions: raise def uninstall_plugin(name=None, **kwargs): """Uninstalls a MySQL Shell plugin. This function uninstall a plugin Args: name (str): The name of the plugin. **kwargs: Optional parameters Keyword Args: interactive (bool): Whether user input is accepted printouts (bool): Whether information should be printed raise_exceptions (bool): Whether exceptions are raised Returns: None or plugin information """ interactive = kwargs.get("interactive", True) printouts = kwargs.get("printouts", True) raise_exceptions = kwargs.get("raise_exceptions", False) try: # Get the list of repositories and their plugins repositories = get_repositories_with_plugins( plugin_filter=Filter.ONLY_INSTALLED, printouts=printouts) if not repositories: return # Get the plugin by name or interactive user selection plugin = get_plugin(repositories, name, interactive) if plugin is None: print("Cancelling operation.") return # Get plugin name and caption name = plugin.get('name') caption = plugin.get('caption') module_name = plugin.get('moduleName') if interactive: prompt = mysqlsh.globals.shell.prompt( f"\nAre you sure you want to uninstall the plugin '{name}'" " [YES/no]: ", { 'defaultValue': 'yes' }).strip().lower() if prompt != 'yes': print("Operation cancelled.") return if printouts: print(f"Uninstalling {caption} ...") # Get plugins_dir plugins_dir = general.get_plugins_dir() # Set plugin directory using the name of the plugin plugin_dir = os.path.join(plugins_dir, module_name) # Delete plugin directory and its contents shutil.rmtree(plugin_dir) if printouts: print(f"{caption} has been uninstalled successfully.\n\n" f"Please restart the shell to unload the plugin.\n") except Exception as e: print(f"ERROR: Could not uninstall the plugin.\n{str(e)}") if raise_exceptions: raise def update_plugin(name=None, **kwargs): """Updates MySQL Shell plugins. This function updates on or all plugins Args: name (str): The name of the plugin. **kwargs: Optional parameters Keyword Args: interactive (bool): Whether user input is accepted raise_exceptions (bool): Whether exceptions are raised Returns: None or plugin information """ interactive = kwargs.get("interactive", True) raise_exceptions = kwargs.get("raise_exceptions", False) try: # Get available repositories including the plugin lists repositories = get_repositories_with_plugins( plugin_filter=Filter.UPDATABLE) if not repositories: return # Build list of plugins and ask the user to confirm when interactive is # set plugins = [] plugin_count = get_plugin_count(repositories=repositories) if plugin_count == 0: print("No updates available.\n") return elif plugin_count == 1: plugin = None if name: plugin = get_plugin(repositories=repositories, name=name, interactive=False) if not plugin: print(f"No update available for plugin '{name}'.") return elif plugin_count == 1: plugin = get_plugin(repositories=repositories) if interactive: caption = plugin.get("caption") prompt = mysqlsh.globals.shell.prompt( f"Are you sure you want to update the '{caption}' " "[YES/no]: ", { 'defaultValue': 'yes' }).strip().lower() if prompt != 'yes': print("Operation cancelled.") return plugins.append(plugin) else: for r in repositories: plugins = r.get("plugins") if not plugins: continue plugins.append(plugins) print("There are updates pending for the following plugins:") for plugin in plugins: print(f" - {plugin.get('caption')}") prompt = mysqlsh.globals.shell.prompt( f"\nAre you sure you want to update these plugins [YES/no]: ", { 'defaultValue': 'yes' }).strip().lower() if prompt != 'yes': print("Operation cancelled.") return # Perform the updates for plugin in plugins: name = plugin.get("name") caption = plugin.get("caption") try: print(f"Updating {caption} ...") install_plugin(name=name, force_install=True, interactive=False, printouts=False, raise_exceptions=True) print(f"\nThe update{'s have' if len(plugins) > 1 else ' has'}" " been completed successfully.\n\n" "Please restart the shell to reload the plugin.\n") except Exception as e: print(f"ERROR: Could not update plugin '{name}'.\n{str(e)}") if raise_exceptions: raise except Exception as e: print(f"ERROR: Could not successfully update all plugins.\n{str(e)}") if raise_exceptions: raise def plugin_details(name=None, **kwargs): """Gives detailed information about a MySQL Shell plugin. Args: name (str): The name of the plugin. **kwargs: Optional parameters Keyword Args: interactive (bool): Whether user input is accepted Returns: None or plugin information """ interactive = kwargs.get("interactive", True) # Get available repositories including the plugin lists repositories = get_repositories_with_plugins() # Get a plugin to print details about plugin = get_plugin(repositories=repositories, name=name, interactive=interactive) if not plugin: return caption = plugin.get('caption', '') description = plugin.get('description', '').replace("\n", f"\n{' ' * 13}") latest_v = plugin.get('latestVersion', '') v_data = get_plugin_version_data(plugin=plugin, version=latest_v) latest_v_dev_stage = v_data.get('developmentStage', '').upper() shell_ver_min = v_data.get('shellVersionMin','') shell_ver_max = v_data.get('shellVersionMax','') versions = plugin.get("versions", []) print(f"{caption}\n{'-' * len(caption)}\n" f"{'Plugin Name:':>20} {plugin.get('name', '')}\n" f"{'Latest Version:':>20} {latest_v}\n" f"{'Dev Stage:':>20} {latest_v_dev_stage}\n" f"{'Min. Shell Version:':>20} {shell_ver_min}\n" f"{'Max. Shell Version:':>20} {shell_ver_max}\n" f"{'Description:':>20} {description}\n" f"{'Available Versions:':>20} {len(versions)}") i = 0 for version in versions: v_str = (f"{version.get('version', '?.?.?')} " f"{version.get('developmentStage', '?').upper()}") v_str = f"{v_str:>18}" out = v_str change_list = version.get("changes") if change_list and len(change_list) > 0: changes = "" for change in change_list: if changes == "": changes += " - " else: changes += f"{' ' * len(v_str)} - " changes += change + "\n" out += changes print(out) i += 1 if interactive and i == 5 and len(versions) > 5: rem = len(versions) - i prompt = mysqlsh.globals.shell.prompt( f"Do you want to print the remaining list of " f"{rem} version{'s' if rem > 1 else ''}? [yes/NO]: ", { 'defaultValue': 'no' }).strip().lower() if prompt != 'yes': break mysqlsh/plugin_manager/registrar.py 0000644 00000074073 15231164450 0013624 0 ustar 00 # Copyright (c) 2020, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 """Plugin Manager used for simplified plugin registration""" import inspect import re from functools import wraps, partial # Callbacks for additional handling on registered plugin # functions should be added here, they should be in the form # def callback(definition) # # Where definition is an instance of FunctionData _registration_callbacks = [] def add_registration_callback(callback): _registration_callbacks.append(callback) def validate_shell_version(min=None, max=None): """ Validates the plugin Shell version requirements for plugin. """ import mysqlsh raw_version = mysqlsh.globals.shell.version shell_version = tuple([int(v) for v in raw_version.split()[1].split('-')[0].split('.')]) # Ensures the plugin can be installed on the current version of the Shell min_version_ok = True if not min is None: min_version = tuple([int(v) for v in min.split('.')]) min_version_ok = shell_version >= min_version max_version_ok = True if not max is None: max_version = tuple([int(v) for v in max.split('.')]) max_version_ok = shell_version <= max_version error = "" if not min_version_ok or not max_version_ok: if min is not None and max is not None: error = f"This plugin requires Shell between versions {min} and {max}." elif min is not None: error = f"This plugin requires at least Shell version {min}." elif max is not None: error = f"This plugin does not work on Shell versions newer than {max}." if len(error) != 0: raise Exception(error) class PluginRegistrar: """Helper class to register a shell plugin. It should be used by calling: register_object which requires: - The object path as it should be seen in the shell. - The python function that will be added as members of the object. - The documentation for the object (brief and details). Examples for object name: - 'cloud': Would register the 'cloud' as a shell global object. - 'cloud.os': Would register the 'os' object as a child of the 'cloud' global object. The name can have any number of parents in the chain, the conditions is that such parent should be already registered. For this reason, the caller may have to do something like: register_object('myGlobal', [], {'brief':'...', 'details':[]}) register_object('myGlobal.myChild', [], {'brief':'...', 'details':[]}) Before calling: register_object('myGlobal.myChild.myGrandChild', [function1, function2], {'brief':'...', 'details':[]}) Any object that is already defined will NOT be re-defined. """ @staticmethod def sphinx_2shell_type(type): """Helper function to translate the sphinx types into the types required by the shell""" if type == "str": return "string" elif type == "int": return "integer" elif type == "dict": return "dictionary" elif type == "list": return "array" else: return type class ItemDoc: """ Simple container for brief and details of the different items""" def __init__(self): self.brief = "" self.details = None class OptionData: """Holds the documentation for a specific option. This object is completely created from parsed docs for an option. TODO: Does not support 'details' for option as sphinx doesn't seem to have a way to specify that, so the details MUST be placed at the function details section (which is OK considering that's what the shell will do when rendering the help data) """ def __init__(self, option_def): self.docs = PluginRegistrar.ItemDoc() self.name = option_def["name"] self.type = option_def.get("type") self.docs.brief = option_def["brief"] self.options = [] self.required = option_def.get("required", False) if "details" in option_def: self.docs.details = option_def["details"] if "options" in option_def: for option in option_def["options"]: self.options.append(PluginRegistrar.OptionData(option)) self.default = None def format_info(self): """Translates the parameter definition to shell required format""" info = { "name": self.name, "brief": self.docs.brief, } if self.type: info["type"] = PluginRegistrar.sphinx_2shell_type(self.type) if self.docs.details is not None: info["details"] = self.docs.details if self.required: info["required"] = True if self.default is not None: info["default"] = self.default if "type" in info.keys() and info["type"] == "dictionary": options = [] for o in self.options: options.append(o.format_info()) info["options"] = options return info class ParameterData: """Holds the documentation for a specific argument. This object is created part from document parsing and part from inspection of the function definition. The function definition provides parameter name, whether it is optional or not and the default value. The documentation provides type, brief and options (when applicable) TODO: Does not support 'details' for option as sphinx doesn't seem to have a way to specify that, so the details MUST be placed at the function details section (which is OK considering that's what the shell will do when rendering the help data) """ def __init__(self, parameter): self.definition = parameter self.docs = PluginRegistrar.ItemDoc() self.type = "" self.options = [] def set_info(self, info): """Fills parameter details that come from documentation""" if "type" in info: self.type = info["type"] self.docs.brief = info["brief"] # Note: this is here even it's not really supported. try: self.docs.details = info["details"] except KeyError: pass try: options = info["options"] for option in options: self.options.append(PluginRegistrar.OptionData(option)) except KeyError: pass def format_info(self): """Translates the parameter definition to shell required format""" info = { "name": self.definition.name, "brief": self.docs.brief, } if self.type: info["type"] = PluginRegistrar.sphinx_2shell_type(self.type) if self.docs.details is not None: info["details"] = self.docs.details if self.definition.default != inspect.Parameter.empty: info["default"] = self.definition.default info["required"] = False if "type" in info.keys() and info["type"] == "dictionary" and len(self.options): options = [] for o in self.options: options.append(o.format_info()) info["options"] = options if self.definition.kind == inspect.Parameter.VAR_KEYWORD: info["required"] = False return info class FunctionData: """Holds the documentation for the function. This object is created part from document parsing and part from inspection of the function definition. The function definition provides parameter information. The documentation provides brief and details. """ def __init__(self, function, fully_qualified_name=None, shell=True, cli=False, web=False): # Get the plugin name by stripping the function name name = None if not fully_qualified_name is None: name_path = fully_qualified_name.split(".") name = name_path[-1] self.function = function self.fully_qualified_name = fully_qualified_name self.name = function.__name__ if name is None else name self.docs = PluginRegistrar.ItemDoc() # These flags indicate where the function should be available self.shell = shell self.cli = cli self.web = web signature = inspect.signature(function) self.parameters = [ PluginRegistrar.ParameterData(p) for p in signature.parameters.values() ] docs = inspect.getdoc(function) if docs: self._parse_docs(docs) def format_info(self): """Translates the parameter definition to shell required format""" info = {"brief": self.docs.brief, "details": self.docs.details} if len(self.parameters): params = [] for param in self.parameters: params.append(param.format_info()) info["parameters"] = params if self.cli: info["cli"] = self.cli return info def _get_doc_section(self, name): """Ensures the indicated section exists and returns it""" try: return self._doc_sections[name] except KeyError: self._doc_sections[name] = [] self._doc_section_list.append(name) return self._doc_sections[name] def _parse_docs(self, docstring): """Main parser function. Retrieves the documentation and splits it in sections, then process the different sections such as: brief, parameters and details. There are different sections that are considered on the parsing: - 'Args:': Contains the parameter definitions. - 'Keyword Args:': Contains parameter definitions for keyword arguments. - 'Allowed options for <dictionary>:': Contains option definitions for a dictionary parameter or option. These options will be consumed during the parsing process. """ self._doc_index = 0 self._doc_lines = [line for line in docstring.split("\n")] self._doc_sections = {} self._doc_section_list = [] # Document splitting logic section_name = "global" section = self._get_doc_section(section_name) global_count = 0 for line in self._doc_lines: if line.endswith(":"): section_name = line section = self._get_doc_section(section_name) elif section_name.startswith("global") or len(line) == 0 or line[0] == " " or line.startswith("* "): section.append(line) else: global_count = global_count + 1 section_name = "global{}".format(global_count) section = self._get_doc_section(section_name) section.append(line) # Removes leading and trailing blank lines from all the sections for name, section in self._doc_sections.items(): name = name # name is not used while section and len(section[0]) == 0: section.pop(0) while section and len(section[-1]) == 0: section.pop(-1) if len(section) == 0: raise Exception(f"Invalid format: section without content: {name}") # Parses the function brief description self._parse_function_brief() # Triggers the parameter parsing and updates the parameter # definitions arg_docs = self._parse_args() if arg_docs: self._set_parameter_docs(arg_docs) # Constructs the function details self._parse_details() def _parse_function_brief(self): """Constructs the function brief from the beginning of the docs and until the first blank line is found""" brief_lines = [] lines = self._doc_sections["global"] for line in lines: if len(line): brief_lines.append(line) else: break count = len(brief_lines) if count: self.docs.brief = " ".join(brief_lines) # Removes any subsequent blank line, rest will be part of the # function details while ( count < len(self._doc_sections["global"]) and len(self._doc_sections["global"][count]) == 0 ): count = count + 1 self._doc_sections["global"] = self._doc_sections["global"][count:] def _parse_args(self): """Parses the function arguments. This function will ensure: - All the parameters are documented - No nonexisting parameters are documented """ args_section = self._doc_sections.pop("Args:", None) arg_docs = [] if len(self.parameters) and args_section is None: raise Exception("Missing arguments documentation") elif args_section is None and len(self.parameters): raise Exception("Unexpected argument documentation") elif args_section: while len(args_section): arg_docs.append(self._parse_parameter_doc(args_section)) return arg_docs def _parse_return_value(self): """Some day it will parse the return section TODO: The shell extension objects do not do any handling of the return value so it's documentation is embedded into the function details (in an ugly way). """ pass def _parse_details(self): """ Creates the details section of the function. When this function is called, the remaining sections will be glued together to create the function details. This means, it is possible to define totally unrelated sections of documentation and they will also be included on the shell docs. """ details = [] paragraph = [] for section_name in self._doc_section_list: section = self._doc_sections.pop(section_name, None) if section: # Sections other than globals need to include the header if not section_name.startswith("global"): details.append("<b>" + section_name + "</b>") # Backup the section index, in case it was not really a # section, i.e. it is the description of a list of items # the bold will be removed section_index = len(details) - 1 for line in section: stripped_line = line.strip() if len(stripped_line) == 0: details.append(" ".join(paragraph)) paragraph.clear() elif line.startswith("* "): # If this is the first thing added to the section, # then it was not a real section but the # description of a list of items. Remove the <b> if section_index == len(details) - 2: details[section_index] = section_name details.append(line.replace("* ", "@li ")) else: paragraph.append(line) if len(paragraph): details.append(" ".join(paragraph)) paragraph.clear() if details: self.docs.details = details def _set_parameter_docs(self, docs): """Fills all the parameter definitions with the information coming from document parsing""" for doc in docs: target_param = None for param in self.parameters: if doc["name"] == param.definition.name: target_param = param if target_param is None: raise Exception( "Parameter does not exist but is documented: {}." "".format(doc) ) else: target_param.set_info(doc) missing = [] for p in self.parameters: if len(p.docs.brief) == 0: missing.append(p.definition.name) if len(missing): raise Exception( "Missing documentation for the following parameters: " "{}".format(", ".join(missing)) ) def _parse_parameter_doc(self, section): """Parses the first parameter/option definition coming on the section. Once it is parsed it will be removed from the section, next call will process the next definition. """ # Parameters are documented as <name> [(<type>)]: <brief> # kwards is documented as **<name>: <brief> info = {} param_line = section.pop(0) param_doc = param_line.strip() brief = [] match = re.match( "^(\\*\\*)?([a-z|A-Z|_][a-z|A-Z|0-9|_]*)(\\s(\\(([a-z|,|\\s]+)\\)))?:\\s(.*)$", param_doc) if match: options_section = None info["name"] = match.group(2) # kwargs definition if match.group(1): info["type"] = "dictionary" options_section = self._doc_sections.pop( "Keyword Args:", None) # data type definition included elif match.group(5): param_options = match.group(5).split(",") info["type"] = PluginRegistrar.sphinx_2shell_type( param_options[0]) if len(param_options) > 1: if param_options[1] == "required": info["required"] = True if info["type"] == "dictionary": options_section = self._doc_sections.pop( "Allowed options for {}:".format(info["name"]), None, ) if options_section: info["options"] = [] while len(options_section): info["options"].append( self._parse_parameter_doc(options_section) ) brief.append(match.group(6)) ident_size = param_line.find(param_doc) while ( len(section) and len(section[0]) > ident_size and section[0][ident_size + 1] == " " ): brief.append(section.pop(0).strip()) info["brief"] = " ".join(brief) else: raise Exception("Invalid parameter documentation: {}" "".format(param_line)) return info def __init__(self): pass def _get_python_name(self, name): return "_".join([x.lower() for x in re.split("([A-Z][a-z0-9_]*)", name) if x]) def register_object(self, name, docs, members=None): try: plugin_obj = self.get_plugin_object(name, docs) if plugin_obj is None: raise ValueError(f"Plugin object {name} was not found.") if members is not None: for member in members: if inspect.isfunction(member): self.register_function(plugin_obj, member) except Exception as e: raise Exception( f"Could not register object '{name}'.\nERROR: {str(e)}" ) def get_plugin_object(self, name, docs): """Get the leaf object in the object hierarchy defined in name. If the leaf object does not exist, it will be created with the provided documentation. If any object in the middle of the chain does not exist, an error will be raised. """ import mysqlsh shell_obj = mysqlsh.globals.shell hierarchy = name.split(".") try: plugin_obj = None if len(hierarchy) > 1: # Get the name of the last plugin object in the # fully qualified name plugin_name = hierarchy[-1] # Loop over all plugins in the hierarchy and ensure they exist parent = mysqlsh.globals for h_name in hierarchy[0:-1]: py_name = h_name # Global objects have the same name in JS/PY, but child # objects are registered as properties in which case naming # convention is followed if parent != mysqlsh.globals: py_name = self._get_python_name(h_name) if py_name in dir(parent): parent = getattr(parent, py_name) else: raise Exception( f"Object {py_name} not found in hierarchy: {name}" ) # Return the plugin object if it already exists py_plugin_name = self._get_python_name(plugin_name) plugin_obj = getattr(parent, py_plugin_name, None) if not plugin_obj: if docs is None: raise ValueError( f"No docs specified for plugin object " f"{name}" ) # If it does not exist yet, create it plugin_obj = shell_obj.create_extension_object() shell_obj.add_extension_object_member( parent, plugin_name, plugin_obj, docs ) else: try: plugin_obj = getattr(mysqlsh.globals, name, None) except KeyError: plugin_obj = shell_obj.create_extension_object() shell_obj.register_global(name, plugin_obj, docs) return plugin_obj except Exception as e: raise Exception( f"Could not get plugin object '{name}'.\nERROR: {str(e)}" ) def register_function( self, plugin_obj, function, fully_qualified_name=None, shell=True, cli=False, web=False ): """Registers a new member into the provided shell extension object""" import mysqlsh shell_obj = mysqlsh.globals.shell if cli and not shell: raise Exception( "The CLI can only be enabled on registered functions.") definition = PluginRegistrar.FunctionData( function, shell=shell, fully_qualified_name=fully_qualified_name, cli=cli, web=web) try: if shell: shell_obj.add_extension_object_member( plugin_obj, definition.name, function, definition.format_info(), ) # On success registration, the function is reported to the callbacks for callback in _registration_callbacks: callback(definition) except Exception as e: raise Exception( f"Could not add function '{definition.name}' " f"using {definition.format_info()}.\nERROR: {str(e)}" ) def register_property(self, property): pass def plugin(cls=None, shell_version_min=None, shell_version_max=None, parent=None): """Decorator to register a class as a Shell extension object This decorator can be used to register a class structure as a Shell extension object. After registering the class it self it will also scan for inner classes and register them as nested extension objects. Args: cls (class): The class structure to register Returns: """ if cls is None: return partial(plugin, shell_version_min=shell_version_min, shell_version_max=shell_version_max, parent=parent) else: try: validate_shell_version(shell_version_min, shell_version_max) plugin_manager = PluginRegistrar() # Use the class name as the plugin name and the DocString as docs and # register the class as Shell plugin object_qualified_name = cls.__name__ if parent is None else parent + "." + cls.__name__ plugin_manager.register_object( object_qualified_name, PluginRegistrar.FunctionData(cls).format_info() ) def register_inner_classes(cls): """Register all inner classes as nested extension objects Args: cls (class): The class containing the subclasses to register """ # Get all inner classes inner_classes = [ inner_class for inner_class in cls.__dict__.values() if inspect.isclass(inner_class) ] # Register those as extension objects for inner_class in inner_classes: plugin_manager.register_object( inner_class.__qualname__, PluginRegistrar.FunctionData(inner_class).format_info(), ) # Recursively also register the inner classes of this class register_inner_classes(inner_class) # Create an instance of the class to also register the plugin functions # initalized in the class's constructor __init__ cls() # Register the inner classes as nested extension objects register_inner_classes(cls) except Exception as e: print( "Could not register plugin object '{0}'.\nERROR: {1}".format( cls.__name__, str(e) ) ) raise @wraps(cls) def wrapper(*args, **kwargs): result = cls(*args, **kwargs) return result return wrapper def plugin_function(fully_qualified_name, plugin_docs=None, shell=True, cli=False, web=False): """Decorator factory to register Shell plugins functions Args: fully_qualified_name (str): The fully qualified name of the function, e.g. cloud.create.mysqlDbSystem plugin_docs (dict): The documentation structure of the plugin. This is only required for the first function that will be registered cli (bool): Defines whether the function should be available in the CLI Returns: The decorator function """ def decorator(function): try: plugin_manager = PluginRegistrar() # Get the plugin name by stripping the function name name_path = fully_qualified_name.split(".") plugin_name = ".".join(name_path[0:-1]) function_name = name_path[-1] # Get the plugin object or create it if it is available yet plugin_obj = plugin_manager.get_plugin_object( plugin_name, plugin_docs ) # register the function plugin_manager.register_function( plugin_obj, function, fully_qualified_name=fully_qualified_name, shell=shell, cli=cli, web=web ) except Exception as e: print( f"Could not register function '{function_name}' as a member " f"of the {plugin_name} plugin object.\nERROR: {str(e)}" ) raise @wraps(function) def wrapper(*args, **kwargs): # TODO investigate if automated handling of certain args is # beneficial result = function(*args, **kwargs) # TODO investigate if transformation of results should be done # when functions are called from the shell prompt return result return wrapper return decorator mysqlsh/plugin_manager/__init__.py 0000644 00000002447 15231164450 0013355 0 ustar 00 # Copyright (c) 2020, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 """The MySQL Shell Plugin Manager""" from .registrar import plugin, plugin_function, validate_shell_version from .general import VERSION mysqlsh/plugin_manager/general.py 0000644 00000014703 15231164450 0013231 0 ustar 00 # Copyright (c) 2020, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 certifi import platform import os import urllib.error import urllib.request import stat import ssl from pathlib import Path # Define plugin version VERSION = "0.0.1" URL_OPEN_TIMEOUT_SECONDS = 10 def get_shell_user_dir(*argv): """Returns the MySQL Shell's user directory The directory will be created if it does not exist yet Args: *argv: The list of directories to be added to the path Returns: The plugin directory path as string """ if "MYSQLSH_USER_CONFIG_HOME" in os.environ: shell_dir = os.environ["MYSQLSH_USER_CONFIG_HOME"] else: home_dir = Path.home() if not home_dir: raise Exception("No home directory set") os_name = platform.system() if os_name == "Windows": shell_dir = os.path.join( home_dir, "AppData", "Roaming", "MySQL", "mysqlsh" ) else: shell_dir = os.path.join(home_dir, ".mysqlsh") # Ensure the path exists try: Path(shell_dir).mkdir(parents=True) except FileExistsError: pass for arg in argv: shell_dir = os.path.join(shell_dir, arg) return shell_dir def get_shell_temp_dir(*argv): """Returns the MySQL Shell's temp directory The directory will be created if it does not exist yet Args: *argv: The list of directories to be added to the path Returns: The temp directory path as string """ temp_dir = get_shell_user_dir("temp") # Ensure the path exists try: Path(temp_dir).mkdir(parents=True) except FileExistsError: pass for arg in argv: temp_dir = os.path.join(temp_dir, arg) return temp_dir def get_plugins_dir(*argv): """Returns the MySQL Shell's plugin directory The directory will be created if it does not exist yet Args: *argv: The list of directories to be added to the path Returns: The plugin directory path as string """ from pathlib import Path import os.path plugins_dir = get_shell_user_dir("plugins") # Ensure the path exists try: Path(plugins_dir).mkdir(parents=True) except FileExistsError: pass for arg in argv: plugins_dir = os.path.join(plugins_dir, arg) return plugins_dir def download_file(url, file_path=None, raise_exceptions=False, ctx=None): """Downloads a file from a given URL Args: url (str): The URL to the text file to download file_path (str): The file to store the downloaded file raise_exceptions (bool): Whether exceptions are raised Returns: The contents of the file decoded to utf-8 if no file is given or True on success or None on failure """ try: os_name = platform.system() if ctx is None and os_name == "Darwin": # NOTE: On this call we require raising exceptions so if it fails in OSX # It still goes and attempt using the certifi-ca below. return download_file(url, file_path=file_path, raise_exceptions=True, ctx=ssl.create_default_context(cafile="/private/etc/ssl/cert.pem")) with urllib.request.urlopen(url, timeout=URL_OPEN_TIMEOUT_SECONDS, context=ctx) as response: data = response.read() if not file_path: return data.decode("utf-8") else: with open(file_path, "wb") as out_file: out_file.write(data) return True except urllib.error.HTTPError as e: if raise_exceptions: raise if e.code == 404: print(f"Could not download file from {url}\nERROR: {str(e)}") return else: raise except urllib.error.URLError as e: # Since the reason might be either text or another exception, we get the # string representation so we can validate the error below. reason = str(e.reason) if 'CERTIFICATE_VERIFY_FAILED' in reason and ctx is None: cafile = install_ssl_certificates() return download_file(url, file_path=file_path, raise_exceptions=raise_exceptions, ctx=ssl.create_default_context(cafile=cafile)) else: print(f"Could not download file from {url}\nERROR: {str(e)}") if raise_exceptions: raise return None def install_ssl_certificates(): STAT_0o775 = ( stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR | stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH ) # Shell CA File Path shell_cafile = os.path.join(get_shell_user_dir(), "certifi-ca.pem") print("Installing SSL certificate...") relpath_to_certifi_cafile = os.path.relpath(certifi.where()) print(" -- removing any existing file or link") try: os.remove(shell_cafile) except FileNotFoundError: pass print(" -- creating symlink to certifi certificate bundle") os.symlink(relpath_to_certifi_cafile, shell_cafile) print(" -- setting permissions") os.chmod(shell_cafile, STAT_0o775) print(" -- update complete") return shell_cafile mysqlsh/plugin_manager/__pycache__/registrar.cpython-39.pyc 0000644 00000051202 15231164450 0020100 0 ustar 00 a kb�g;x � @ s` d Z ddlZddlZddlmZmZ g Zdd� Zddd�ZG dd � d �Z dd d �Z ddd�Z dS )z6Plugin Manager used for simplified plugin registration� N)�wraps�partialc C s t �| � d S �N)�_registration_callbacks�append)�callback� r �D/usr/lib/mysqlsh/python-packages/mysqlsh/plugin_manager/registrar.py�add_registration_callback% s r c C s ddl }|jjj}tdd� |�� d �d�d �d�D ��}d}| durhtd d� | �d�D ��}||k}d}|dur�td d� |�d�D ��}||k}d } |r�|s�| dur�|dur�d | � d |� d�} n*| dur�d| � d�} n|dur�d|� d�} t| �dk�rt| ��dS )zE Validates the plugin Shell version requirements for plugin. r Nc S s g | ] }t |��qS r ��int��.0�vr r r � <listcomp>0 � z*validate_shell_version.<locals>.<listcomp>� �-�.Tc S s g | ] }t |��qS r r r r r r r 5 r c S s g | ] }t |��qS r r r r r r r : r � z,This plugin requires Shell between versions z and z,This plugin requires at least Shell version z7This plugin does not work on Shell versions newer than )�mysqlsh�globals�shell�version�tuple�split�len� Exception) �min�maxr Z raw_versionZ shell_versionZmin_version_okZ min_versionZmax_version_okZ max_version�errorr r r �validate_shell_version) s* * r! c @ s� e Zd ZdZedd� �ZG dd� d�ZG dd� d�ZG dd � d �ZG d d � d �Z d d � Z dd� Z ddd�Z dd� Z d dd�Zdd� ZdS ) �PluginRegistrara Helper class to register a shell plugin. It should be used by calling: register_object which requires: - The object path as it should be seen in the shell. - The python function that will be added as members of the object. - The documentation for the object (brief and details). Examples for object name: - 'cloud': Would register the 'cloud' as a shell global object. - 'cloud.os': Would register the 'os' object as a child of the 'cloud' global object. The name can have any number of parents in the chain, the conditions is that such parent should be already registered. For this reason, the caller may have to do something like: register_object('myGlobal', [], {'brief':'...', 'details':[]}) register_object('myGlobal.myChild', [], {'brief':'...', 'details':[]}) Before calling: register_object('myGlobal.myChild.myGrandChild', [function1, function2], {'brief':'...', 'details':[]}) Any object that is already defined will NOT be re-defined. c C s8 | dkr dS | dkrdS | dkr$dS | dkr0dS | S d S ) zZHelper function to translate the sphinx types into the types required by the shell�str�stringr Zinteger�dict� dictionary�listZarrayNr )�typer r r �sphinx_2shell_typed s z"PluginRegistrar.sphinx_2shell_typec @ s e Zd ZdZdd� ZdS )zPluginRegistrar.ItemDocz> Simple container for brief and details of the different itemsc C s d| _ d | _d S �Nr ��brief�details��selfr r r �__init__v s z PluginRegistrar.ItemDoc.__init__N)�__name__� __module__� __qualname__�__doc__r0 r r r r �ItemDocs s r5 c @ s e Zd ZdZdd� Zdd� ZdS )zPluginRegistrar.OptionDataa� Holds the documentation for a specific option. This object is completely created from parsed docs for an option. TODO: Does not support 'details' for option as sphinx doesn't seem to have a way to specify that, so the details MUST be placed at the function details section (which is OK considering that's what the shell will do when rendering the help data) c C s� t �� | _|d | _|�d�| _|d | j_g | _|�dd�| _d|v rT|d | j_ d|v r||d D ]}| j� t � |�� qdd | _ d S )N�namer( r, �requiredFr- �options) r" r5 �docsr6 �getr( r, r8 r7 r- r � OptionData�default)r/ Z option_def�optionr r r r0 � s z#PluginRegistrar.OptionData.__init__c C s� | j | jjd�}| jr&t�| j�|d< | jjdur>| jj|d< | jrLd|d< | jdur`| j|d< d|� � v r�|d dkr�g }| j D ]}|� |� � � q�||d < |S ) �<Translates the parameter definition to shell required format�r6 r, r( Nr- Tr7 r< r& r8 ) r6 r9 r, r( r" r) r- r7 r< �keysr8 r � format_info�r/ �infor8 �or r r rA � s" � z&PluginRegistrar.OptionData.format_infoN)r1 r2 r3 r4 r0 rA r r r r r; z s r; c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) z PluginRegistrar.ParameterDataa� Holds the documentation for a specific argument. This object is created part from document parsing and part from inspection of the function definition. The function definition provides parameter name, whether it is optional or not and the default value. The documentation provides type, brief and options (when applicable) TODO: Does not support 'details' for option as sphinx doesn't seem to have a way to specify that, so the details MUST be placed at the function details section (which is OK considering that's what the shell will do when rendering the help data) c C s || _ t�� | _d| _g | _d S r* )� definitionr" r5 r9 r( r8 )r/ Z parameterr r r r0 � s z&PluginRegistrar.ParameterData.__init__c C s� d|v r|d | _ |d | j_z|d | j_W n ty@ Y n0 z(|d }|D ]}| j�t�|�� qPW n ty| Y n0 dS )z4Fills parameter details that come from documentationr( r, r- r8 N) r( r9 r, r- �KeyErrorr8 r r" r; )r/ rC r8 r= r r r �set_info� s z&PluginRegistrar.ParameterData.set_infoc C s� | j j| jjd�}| jr(t�| j�|d< | jjdur@| jj|d< | j jt j j krd| j j|d< d|d< d|� � v r�|d dkr�t | j�r�g }| jD ]}|�|�� � q�||d < | j jt j jkr�d|d< |S ) r> r? r( Nr- r< Fr7 r& r8 )rE r6 r9 r, r( r" r) r- r< �inspectZ Parameter�emptyr@ r r8 r rA �kindZ VAR_KEYWORDrB r r r rA � s$ � " z)PluginRegistrar.ParameterData.format_infoN)r1 r2 r3 r4 r0 rG rA r r r r � ParameterData� s rK c @ sb e Zd ZdZddd�Zdd� Zd d � Zd d � Zd d� Zdd� Z dd� Z dd� Z dd� Z dd� Z dS )z PluginRegistrar.FunctionDataa# Holds the documentation for the function. This object is created part from document parsing and part from inspection of the function definition. The function definition provides parameter information. The documentation provides brief and details. NTFc C s� d }|d ur |� d�}|d }|| _|| _|d u r8|jn|| _t�� | _|| _|| _ || _ t � |�}dd� |j �� D �| _ t �|�} | r�| �| � d S )Nr ���c S s g | ]}t �|��qS r )r" rK )r �pr r r r s �z9PluginRegistrar.FunctionData.__init__.<locals>.<listcomp>)r �function�fully_qualified_namer1 r6 r" r5 r9 r �cli�webrH � signature� parameters�valuesZgetdoc� _parse_docs) r/ rN rO r rP rQ r6 � name_pathrR r9 r r r r0 s$ � z%PluginRegistrar.FunctionData.__init__c C sV | j j| j jd�}t| j�rBg }| jD ]}|�|�� � q&||d< | jrR| j|d< |S )r> r+ rS rP )r9 r, r- r rS r rA rP )r/ rC Zparams�paramr r r rA " s z(PluginRegistrar.FunctionData.format_infoc C sD z | j | W S ty> g | j |< | j�|� | j | Y S 0 dS )z3Ensures the indicated section exists and returns itN)� _doc_sectionsrF �_doc_section_listr �r/ r6 r r r �_get_doc_section2 s z-PluginRegistrar.FunctionData._get_doc_sectionc C s` d| _ dd� |�d�D �| _i | _g | _d}| �|�}d}| jD ]|}|�d�r^|}| �|�}q@|�d�s�t|�dks�|d dks�|�d�r�|� |� q@|d }d � |�}| �|�}|� |� q@| j� � D ]j\}}|}|r�t|d �dkr�|� d� q�|�rt|d �dk�r|� d � q�t|�dkr�t d |� ���q�| �� | �� }|�rT| �|� | �� d S )a� Main parser function. Retrieves the documentation and splits it in sections, then process the different sections such as: brief, parameters and details. There are different sections that are considered on the parsing: - 'Args:': Contains the parameter definitions. - 'Keyword Args:': Contains parameter definitions for keyword arguments. - 'Allowed options for <dictionary>:': Contains option definitions for a dictionary parameter or option. These options will be consumed during the parsing process. r c S s g | ]}|�qS r r )r �liner r r r L r z<PluginRegistrar.FunctionData._parse_docs.<locals>.<listcomp>� �global�:� �* r zglobal{}rL z)Invalid format: section without content: N)Z _doc_indexr Z _doc_linesrX rY r[ �endswith� startswithr r �format�items�popr �_parse_function_brief� _parse_args�_set_parameter_docs�_parse_details)r/ Z docstring� section_name�sectionZ global_countr\ r6 �arg_docsr r r rU ; s< , z(PluginRegistrar.FunctionData._parse_docsc C s� g }| j d }|D ] }t|�r*|�|� q q0qt|�}|rJd�|�| j_|t| j d �k r|t| j d | �dkr||d }qJ| j d |d� | j d< dS )zpConstructs the function brief from the beginning of the docs and until the first blank line is foundr^ r` r r N)rX r r �joinr9 r, )r/ Z brief_lines�linesr\ �countr r r rg y s �� z2PluginRegistrar.FunctionData._parse_function_briefc C sl | j �dd�}g }t| j�r.|du r.td��n:|du rJt| j�rJtd��n |rht|�rh|�| �|�� qN|S )z�Parses the function arguments. This function will ensure: - All the parameters are documented - No nonexisting parameters are documented zArgs:Nz Missing arguments documentationz!Unexpected argument documentation)rX rf r rS r r �_parse_parameter_doc)r/ Z args_sectionrm r r r rh � s z(PluginRegistrar.FunctionData._parse_argsc C s dS )z�Some day it will parse the return section TODO: The shell extension objects do not do any handling of the return value so it's documentation is embedded into the function details (in an ugly way). Nr r. r r r �_parse_return_value� s z0PluginRegistrar.FunctionData._parse_return_valuec C s� g }g }| j D ]�}| j�|d�}|r|�d�s@|�d| d � t|�d }|D ]r}|�� }t|�dkr�|�d�|�� |�� qP|�d�r�|t|�d kr�|||< |�|� dd �� qP|�|� qPt|�r|�d�|�� |�� q|r�|| j _ dS ) aZ Creates the details section of the function. When this function is called, the remaining sections will be glued together to create the function details. This means, it is possible to define totally unrelated sections of documentation and they will also be included on the shell docs. Nr^ z<b>z</b>r r r` ra � z@li ) rY rX rf rc r r �striprn �clear�replacer9 r- )r/ r- Z paragraphrk rl Z section_indexr\ Z stripped_liner r r rj � s. z+PluginRegistrar.FunctionData._parse_detailsc C s� |D ]J}d}| j D ]}|d |jjkr|}q|du rDtd�|���q|�|� qg }| j D ]"}t|jj�dkrZ|� |jj� qZt|�r�td�d� |����dS )zaFills all the parameter definitions with the information coming from document parsingNr6 z/Parameter does not exist but is documented: {}.r z6Missing documentation for the following parameters: {}z, ) rS rE r6 r rd rG r r9 r, r rn )r/ r9 �docZ target_paramrW ZmissingrM r r r ri � s, �� ��z0PluginRegistrar.FunctionData._set_parameter_docsc C s� i }|� d�}|�� }g }t�d|�}|�rnd}|�d�|d< |�d�r`d|d< | j� d d�}np|�d �r�|�d ��d �}t�|d �|d< t |�dkr�|d d kr�d |d < |d dkr�| j� d� |d �d�}|r�g |d< t |�r�|d � | � |�� q�|� |�d�� |� |�} t |��r^t |d �| k�r^|d | d dk�r^|� |� d��� � �qd�|�|d< ntd� |���|S )z�Parses the first parameter/option definition coming on the section. Once it is parsed it will be removed from the section, next call will process the next definition. r zE^(\*\*)?([a-z|A-Z|_][a-z|A-Z|0-9|_]*)(\s(\(([a-z|,|\s]+)\)))?:\s(.*)$Nrs r6 r r& r( z Keyword Args:� �,r7 TzAllowed options for {}:r8 � r` r, z#Invalid parameter documentation: {})rf rt �re�match�grouprX r r" r) r rd r rq �findrn r ) r/ rl rC Z param_lineZ param_docr, r| Zoptions_sectionZ param_optionsZ ident_sizer r r rq � s` � � � �� ����z1PluginRegistrar.FunctionData._parse_parameter_doc)NTFF)r1 r2 r3 r4 r0 rA r[ rU rg rh rr rj ri rq r r r r � FunctionData� s � >, r c C s d S r r r. r r r r0 : s zPluginRegistrar.__init__c C s d� dd� t�d|�D ��S )N�_c S s g | ]}|r|� � �qS r )�lower)r �xr r r r > r z4PluginRegistrar._get_python_name.<locals>.<listcomp>z([A-Z][a-z0-9_]*))rn r{ r rZ r r r �_get_python_name= s z PluginRegistrar._get_python_nameNc C s� zP| � ||�}|d u r&td|� d���|d urN|D ]}t�|�r2| �||� q2W n< ty� } z$td|� dt|�� ���W Y d }~n d }~0 0 d S )NzPlugin object z was not found.zCould not register object '� '. ERROR: )�get_plugin_object� ValueErrorrH Z isfunction�register_functionr r# )r/ r6 r9 Zmembers� plugin_objZmember�er r r �register_object@ s �z PluginRegistrar.register_objectc C sb ddl }|jj}|�d�}�zd}t|�dkr�|d }|j}|dd� D ]H} | } ||jkrd| �| �} | t|�v r|t|| �}qHtd| � d|� ���qH| �|�} t|| d�}|s�|du r�t d|� ���|� � }|� ||||� n>zt|j|d�}W n* t �y |� � }|� |||� Y n0 |W S t�y\ } z$td |� d t| �� ���W Y d} ~ n d} ~ 0 0 dS ) a Get the leaf object in the object hierarchy defined in name. If the leaf object does not exist, it will be created with the provided documentation. If any object in the middle of the chain does not exist, an error will be raised. r Nr r rL zObject z not found in hierarchy: z$No docs specified for plugin object z Could not get plugin object 'r� )r r r r r r� �dir�getattrr r� Zcreate_extension_object�add_extension_object_memberrF Zregister_globalr# ) r/ r6 r9 r � shell_objZ hierarchyr� � plugin_name�parentZh_nameZpy_nameZpy_plugin_namer� r r r r� O sL � ���z!PluginRegistrar.get_plugin_objectTFc C s� ddl }|jj}|r |s td��tj|||||d�} z0|rP|�|| j|| �� � t D ] } | | � qTW nH ty� } z0td| j� d| �� � dt | �� ���W Y d} ~ n d} ~ 0 0 dS )z?Registers a new member into the provided shell extension objectr Nz4The CLI can only be enabled on registered functions.)r rO rP rQ zCould not add function 'z' using z . ERROR: ) r r r r r" r r� r6 rA r r# ) r/ r� rN rO r rP rQ r r� rE r r� r r r r� � s< ��� ���z!PluginRegistrar.register_functionc C s d S r r )r/ �propertyr r r �register_property� s z!PluginRegistrar.register_property)N)NTFF)r1 r2 r3 r4 � staticmethodr) r5 r; rK r r0 r� r� r� r� r� r r r r r" H s& 7F E A � (r" c s� � du rt t|||d�S z\t||� t� �|du r8� jn |d � j }��|t�� ��� � ��fdd���� � W n< ty� } z$t d� � jt |��� � W Y d}~n d}~0 0 t � �� fdd��}|S dS ) ag Decorator to register a class as a Shell extension object This decorator can be used to register a class structure as a Shell extension object. After registering the class it self it will also scan for inner classes and register them as nested extension objects. Args: cls (class): The class structure to register Returns: N)�shell_version_min�shell_version_maxr� r c sH dd� | j �� D �}|D ]$}� �|jt�|��� � �|� q| � dS )z�Register all inner classes as nested extension objects Args: cls (class): The class containing the subclasses to register c S s g | ]}t �|�r|�qS r )rH Zisclass)r � inner_classr r r r � s �z:plugin.<locals>.register_inner_classes.<locals>.<listcomp>N)�__dict__rT r� r3 r" r rA )�clsZ inner_classesr� )�plugin_manager�register_inner_classesr r r� � s � � z&plugin.<locals>.register_inner_classesz2Could not register plugin object '{0}'. ERROR: {1}c s � | i |��}|S r r ��args�kwargs�result)r� r r �wrapper� s zplugin.<locals>.wrapper) r �pluginr! r" r1 r� r rA r �printrd r# r )r� r� r� r� Zobject_qualified_namer� r� r )r� r� r� r r� � s* � ��r� TFc s � ����fdd�}|S )a� Decorator factory to register Shell plugins functions Args: fully_qualified_name (str): The fully qualified name of the function, e.g. cloud.create.mysqlDbSystem plugin_docs (dict): The documentation structure of the plugin. This is only required for the first function that will be registered cli (bool): Defines whether the function should be available in the CLI Returns: The decorator function c s� zPt � }��d�}d�|dd� �}|d }|�|��}|j|� ����d� W nD ty� } z,td|� d|� dt|�� �� � W Y d }~n d }~0 0 t� �� fdd ��}|S ) Nr r rL )rO r rP rQ z Could not register function 'z' as a member of the z plugin object. ERROR: c s � | i |��}|S r r r� �rN r r r� 3 s z3plugin_function.<locals>.decorator.<locals>.wrapper) r" r rn r� r� r r� r# r )rN r� rV r� Z function_namer� r� r� �rP rO � plugin_docsr rQ r� r � decorator s6 �� ���z"plugin_function.<locals>.decoratorr )rO r� r rP rQ r� r r� r �plugin_function s (r� )NN)NNNN)NTFF) r4 rH r{ � functoolsr r r r r! r" r� r� r r r r �<module> s" u K � mysqlsh/plugin_manager/__pycache__/__init__.cpython-39.pyc 0000644 00000000527 15231164450 0017641 0 ustar 00 a kb�g' � @ s( d Z ddlmZmZmZ ddlmZ dS )z The MySQL Shell Plugin Manager� )�plugin�plugin_function�validate_shell_version)�VERSIONN)�__doc__Z registrarr r r Zgeneralr � r r �C/usr/lib/mysqlsh/python-packages/mysqlsh/plugin_manager/__init__.py�<module> s mysqlsh/plugin_manager/__pycache__/repositories.cpython-39.pyc 0000644 00000022550 15231164450 0020631 0 ustar 00 a kb�gk6 � @ s� d dl Zd dlZd dlZd dlmZmZ ddlmZ ddddd �gZ d d � Z d d d�Z d dd�Z ed�dd� �Z ed�d dd��Zdd� Zed�dd� �ZdS ) � N)�plugin�plugin_function� )�generalz&Official MySQL Shell Plugin RepositoryzRThe official MySQL Shell Plugin Repository maintained by the MySQL Team at Oracle.z6https://cdn.mysql.com/mysql_shell/plugins_manifest.zipz*manifest/mysql-shell-plugins-manifest.json)�name� description�urlZ manifestPathc C sX d}d}| D ]F}||d�d|� dd�� d|� dd �� d|� d d�� d �7 }|d7 }q |S ) z�Returns a formatted list of repositories. Args: repositories (list): A list of repositories. Returns: The formated list as string � r z>4� r z??z r �-r z ��get)� repositories�out�i�r� r �G/usr/lib/mysqlsh/python-packages/mysqlsh/plugin_manager/repositories.py�format_repository_listing. s � � r Fc C s t j�t�� d�}t j�|�� r�z@t|d�� }t�|�� d�}W d � n1 sR0 Y |W S tj y� } z>t d|� dt |j �� d|j� d|j� �� | r�� W Y d}~dS d}~0 t� y� } z*t d |� dt |�� �� | r� W Y d}~n d}~0 0 ng S dS ) z�Fetches the registered user repositories Args: raise_exceptions (bool): Whether exceptions are raised Returns: The registered repositories as list �plugin-repositories.jsonr r Nz"ERROR: Could not parse JSON file '�'. �, line: �, col: zERROR: Error reading file ')�os�path�joinr �get_shell_user_dir�isfile�open�json�loadr �JSONDecodeError�print�str�msg�lineno�colno�OSError)�raise_exceptions�user_repo_file_path�user_repo_file� user_repos�er r r �get_user_repositoriesB s8 � .���� �r- c C s� dd| d�}t j�t�� d�}z@t|d�� }tj||dd� W d � n1 sP0 Y W d S ty� } z,t d |� d t |�� �� |r�� W Y d}~d S d}~0 0 dS ) z�Stores the registered user repositories Args: user_repos (list): The list of raise_exceptions (bool): Whether exceptions are raised Returns: True on success z MySQL Shell Plugin Repositoriesz0.0.1)ZfileType�versionr r �w� )�indentNTz ERROR: Error writing to file 'r F) r r r r r r r �dumpr' r" r# )r+ r( Zuser_repo_file_contentr) r* r, r r r �set_user_repositoriesc s$ � � . �r3 zplugins.repositories.listc K sz | � dd�}| � dd�}t�� }t|d�}|r8|�|� |s@|S tdt|d�� dt|�� d t|�d krjd nd � d �� dS )a� Lists all registered plugin repositories This function will list all registered MySQL Shell plugin repositories. To add a repository use the plugins.repositories.add() function. Args: **kwargs: Optional parameters Keyword Args: return_formatted (bool): If set to true, a list object is returned. raise_exceptions (bool): Whether exceptions are raised Returns: None or a list of dicts representing the plugins �return_formattedTr( F�r( z-Registered MySQL Shell Plugin Repositories. )r z Total of z repositorr �yZiesz. N)r �DEFAULT_PLUGIN_REPOSITORIES�copyr- �extendr" r �len)�kwargsr4 r( Zreposr+ r r r �get_plugin_repositories� s ���r< zplugins.repositories.addc K s |� dd�}|� dd�}| s@|r@td� tjj�dddi��� } | sPtd� d S | �� �d �r�| � d �}t |�d kr�td � d S |d }t |�dkr�d}n|d }d|� d |� d�} nH| �d�s�| �d�s�d| � �} | � d�s�| � d �r�| � d�} n | � d�} �z�t � | �}|�sW d S t�|�}|� d�} | � d�} | �sJtd� W d S | � d�} | �shtd� W d S |� d �} | �r�t | �d k�r�td � W d S td � td � d!| � d"d#� d!| � d$| � d%� � | D ] } td&| � d'�� d"�� �q�tjj�d(| � d)�dd*i��� �� }|d+k�r td,� W d S td-� t|d.�}|d u �rBW d S |D ]6}|� d/��� | �� k�rFtd0| � d1�� W d S �qFtd2| � d3�� |�| | | d4�� t|��r�td5| � d6�� W nV tj�y } z:td7t|j�� d8|j� d9|j� �� |�r�� W Y d }~d S d }~0 0 d S ):ae Adds a new plugin repository By calling this function a new plugin repository can be added. The url parameter allows for shortcuts. 'domain.com' expands to https://domain.com/mysql-shell-plugins-manifest.json 'domain.com/plugins' expands to https://domain.com/plugins/mysql-shell-plugins-manifest.json 'github/username' expands to https://raw.githubusercontent.com/username/mysql-shell-plugins/master/mysql-shell-plugins-manifest.json 'github/username/repo' expands to https://raw.githubusercontent.com/username/repo/master/mysql-shell-plugins-manifest.json Args: url (str): The url of a MySQL Shell plugins repository. **kwargs: Optional parameters Keyword Args: interactive (bool): Whether user input is accepted raise_exceptions (bool): Whether exceptions are raised Returns: None � interactiveTr( z�To add a new MySQL Shell plugin repository the URL of the repository manifest file needs to be specified. Examples: domain.com domain.com/plugins github/username github/username/repository z/Please enter the URL of the plugin repository: Z defaultValuer z"No URL given. Cancelling OperationNzgithub/�/r z#A github user needs to be specified� zmysql-shell-pluginsz"https://raw.githubusercontent.com/z)/master/mysql-shell-plugins-manifest.jsonzhttps://zhttp://z.jsonz!mysql-shell-plugins-manifest.jsonz"/mysql-shell-plugins-manifest.json� repositoryr z0The given repository misses the 'name' property.r z7The given repository misses the 'description' property.�pluginsr zAThe repository does not contain any plugins. Operation cancelled.a� WARNING: You are about to add an external MySQL Shell plugin repository. External plugin repositories and their plugins complement the functionality of MySQL Shell and can contain system level software that could be potentially harmful to your system. Please review the description below and only proceed if you have obtained the external plugin repository URL from a trusted source. Oracle and its affiliates cannot be held responsible for any potential harm caused by using plugins from external sources. z Repository :r � z Description:z URL: z1 The repository contains the following plugins: z - Zcaptionz-Are you sure you want to add the repository 'z ' [yes/NO]: �no�yeszOperation cancelled.z%Fetching current user repositories...r5 r �The repository 'z' has already been added. zAdding repository '�'...)r r r z Repository 'z' successfully added. z"ERROR: Could not parse JSON file. r r )r r" �mysqlsh�globals�shell�prompt�strip�lower� startswith�splitr: �endswithr � download_filer �loadsr- �appendr3 r! r# r$ r% r& )r r; r= r( Z github_strZ github_userZ github_repoZmanifestZmanifest_contentr@ r r rA r rJ r+ r r, r r r �add_plugin_repository� s� �� � ��� �� � ��rS c C s@ d}d}|D ]"}|� d�| kr&d} q0|d7 }q |r8|S d S d S )Nr Fr Tr r )r r+ �nr�foundr r r r �search_user_repoD s rV zplugins.repositories.removec K s� | � d�}| � dd�}| � dd�}t|d�}|s<td� dS d}|rft||�}|du rftd |� d �� |du �r<|�r<td t|�� �� |du �r<tjj�d �� � }|�r<z`z.t |�}|d kr�|t |�kr�|d }nt �W n, t �y t||�}|du �rt �Y n0 W n( t t f�y2 td|� d�� Y n0 q��q<q�|du �rPtd� n8|| � dd�}td|� d�� ||= t|��r�td� dS )a� Removes a registered plugin repository This function will remove a shell plugin repository previously registered. Args: **kwargs: Optional parameters Keyword Args: url (str): The url of a MySQL Shell plugins repository. interactive (bool): Whether user input is accepted raise_exceptions (bool): Whether exceptions are raised Returns: None r r= Tr( Fr5 z6No custom MySQL Shell plugin repositories registered. Nz No user repository matches '�'z Removing a plugin repository. z/Please enter the index or URL of a repository: r r rE zJ' was not found. Please try again or leave empty to cancel the operation. z,No valid URL provided. Cancelling operation.r z Removing repository 'rF z!Repository successfully removed. )r r- r" rV r rG rH rI rJ rK �intr: � IndexError� ValueErrorr3 )r; r r= r( r+ Z repo_indexrJ rT r r r �remove_plugin_repositoryT sR � � r[ )F)F)N)�os.pathr r rG �mysqlsh.plugin_managerr r r r r7 r r- r3 r< rS rV r[ r r r r �<module> s( � ! % mysqlsh/plugin_manager/__pycache__/general.cpython-39.pyc 0000644 00000010412 15231164450 0017511 0 ustar 00 a kb�g� � @ sz d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlm Z dZ dZ dd� Z dd� Z d d � Zdd d �Zdd� ZdS )� N��Pathz0.0.1� c G s� dt jv rt jd }nHt�� }|s*td��t�� }|dkrPt j�|dddd�}nt j�|d�}zt|�j d d � W n t y� Y n0 | D ]}t j�||�}q�|S ) z�Returns the MySQL Shell's user directory The directory will be created if it does not exist yet Args: *argv: The list of directories to be added to the path Returns: The plugin directory path as string ZMYSQLSH_USER_CONFIG_HOMEzNo home directory setZWindowsZAppDataZRoamingZMySQL�mysqlshz.mysqlshT��parents) �os�environr �home� Exception�platform�system�path�join�mkdir�FileExistsError)�argvZ shell_dirZhome_dir�os_name�arg� r �B/usr/lib/mysqlsh/python-packages/mysqlsh/plugin_manager/general.py�get_shell_user_dir( s$ � r c G sL t d�}zt|�jdd� W n ty. Y n0 | D ]}tj�||�}q4|S )z�Returns the MySQL Shell's temp directory The directory will be created if it does not exist yet Args: *argv: The list of directories to be added to the path Returns: The temp directory path as string ZtempTr )r r r r r r r )r Ztemp_dirr r r r �get_shell_temp_dirM s r c G s` ddl m} ddl}td�}z||�jdd� W n tyB Y n0 | D ]}|j�||�}qH|S )z�Returns the MySQL Shell's plugin directory The directory will be created if it does not exist yet Args: *argv: The list of directories to be added to the path Returns: The plugin directory path as string r r NZpluginsTr )�pathlibr Zos.pathr r r r r )r r r Z plugins_dirr r r r �get_plugins_dirf s r Fc C s� z�t �� }|du r4|dkr4t| |dtjdd�d�W S tjj| t|d��}|� � }W d � n1 sd0 Y |s~|� d�W S t |d ��}|� |� W d � n1 s�0 Y W dS W n� tj j�y } zF|rւ |jd k�rtd | � d t|�� �� W Y d}~dS � W Y d}~n�d}~0 tj j�y� } ztt|j�} d | v �rz|du �rzt� } t| ||tj| d�d�W Y d}~S td | � d t|�� �� |�r�� W Y d}~n d}~0 0 dS )aj Downloads a file from a given URL Args: url (str): The URL to the text file to download file_path (str): The file to store the downloaded file raise_exceptions (bool): Whether exceptions are raised Returns: The contents of the file decoded to utf-8 if no file is given or True on success or None on failure NZDarwinTz/private/etc/ssl/cert.pem)�cafile)� file_path�raise_exceptions�ctx)Ztimeout�contextzutf-8�wbi� z Could not download file from z ERROR: ZCERTIFICATE_VERIFY_FAILED)r r � download_file�sslZcreate_default_context�urllibZrequestZurlopen�URL_OPEN_TIMEOUT_SECONDS�read�decode�open�write�errorZ HTTPError�code�print�strZURLError�reason�install_ssl_certificates) Zurlr r r r Zresponse�dataZout_file�er- r r r r r! � sF �& ( �r! c C s� t jt jB t jB t jB t jB t jB t jB t jB } t j � t � d�}t d� t j �t�� �}t d� zt �|� W n ty� Y n0 t d� t �||� t d� t �|| � t d� |S )Nzcertifi-ca.pemz Installing SSL certificate...z& -- removing any existing file or linkz2 -- creating symlink to certifi certificate bundlez -- setting permissionsz -- update complete)�stat�S_IRUSR�S_IWUSR�S_IXUSR�S_IRGRP�S_IWGRP�S_IXGRP�S_IROTH�S_IXOTHr r r r r+ �relpath�certifi�where�remove�FileNotFoundError�symlink�chmod)Z STAT_0o775Z shell_cafileZrelpath_to_certifi_cafiler r r r. � s< �������� r. )NFN)r; r r Z urllib.errorr# Zurllib.requestr1 r" r r �VERSIONr$ r r r r! r. r r r r �<module> s % : mysqlsh/__init__.py 0000644 00000004317 15231164450 0010363 0 ustar 00 # Copyright (c) 2020, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 class Error(Exception): def __init__(self, code, msg=None): if msg is None: msg = code code = None self.code = code self.msg = msg self.args = (code, msg) def __str__(self): if not self.code: msg = "Shell Error: %s" % (self.msg, ) else: msg = "Shell Error (%s): %s" % (self.code, self.msg) return msg class DBError(Error): def __init__(self, code, msg, sqlstate=None): super().__init__(code, msg) self.sqlstate = sqlstate def __str__(self): return "MySQL Error (%s): %s" % (self.code, self.msg) class ShellGlobals(object): def __setattr__(self, name, value): self.__dict__[name] = value def __delattr__(self, name): del self.__dict__[name] def __getattr__(self, name): # backwards compatibility if name == "mysql": return mysql elif name == "mysqlx": return mysqlx return self.__dict__[name] globals = ShellGlobals() del ShellGlobals mysqlsh/__pycache__/__init__.cpython-39.pyc 0000644 00000003474 15231164450 0014655 0 ustar 00 a kb�g� � @ s< G d d� de �ZG dd� de�ZG dd� de�Ze� Z[dS )c @ s e Zd Zddd�Zdd� ZdS )�ErrorNc C s* |d u r|}d }|| _ || _||f| _d S �N)�code�msg�args)�selfr r � r �4/usr/lib/mysqlsh/python-packages/mysqlsh/__init__.py�__init__ s zError.__init__c C s( | j sd| jf }nd| j | jf }|S )NzShell Error: %szShell Error (%s): %s�r r )r r r r r �__str__! s z Error.__str__)N)�__name__� __module__� __qualname__r r r r r r r s r c s& e Zd Zd� fdd� Zdd� Z� ZS )�DBErrorNc s t � �||� || _d S r )�superr �sqlstate)r r r r �� __class__r r r * s zDBError.__init__c C s d| j | jf S )NzMySQL Error (%s): %sr )r r r r r . s zDBError.__str__)N)r r r r r � __classcell__r r r r r ) s r c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )� ShellGlobalsc C s || j |< d S r ��__dict__)r �name�valuer r r � __setattr__3 s zShellGlobals.__setattr__c C s | j |= d S r r �r r r r r � __delattr__6 s zShellGlobals.__delattr__c C s" |dkr t S |dkrtS | j| S )N�mysql�mysqlx)r r r r r r r � __getattr__9 s zShellGlobals.__getattr__N)r r r r r r r r r r r 2 s r N)� Exceptionr r �objectr �globalsr r r r �<module> s mysql_gadgets/command/sandbox.py 0000644 00000201764 15231164450 0013050 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """This file contains the functionality to create a sandbox of MySQL servers. """ from __future__ import print_function import errno import getpass import logging import os import time import shutil import subprocess import sys try: from ctypes import cdll _CURRENT_ANSI_CODE_PAGE = cdll.kernel32.GetACP() except: _CURRENT_ANSI_CODE_PAGE = 1252 from mysql_gadgets.common import tools, server from mysql_gadgets.common.constants import PATH_ENV_VAR from mysql_gadgets.common.config_parser import (create_option_file, MySQLOptionsParser, option_list_to_dictionary) from mysql_gadgets.common.logger import CustomLevelLogger from mysql_gadgets import exceptions, MIN_MYSQL_VERSION, MAX_MYSQL_VERSION if sys.version_info[0] >= 3: unicode = str # get module logger logging.setLoggerClass(CustomLevelLogger) _LOGGER = logging.getLogger(__name__) _CREATE_SANDBOX_CMD = (u"{mysqld_path} --defaults-file={config_file} " u"--initialize-insecure") _START_SERVER_CMD = u"{mysqld_path} --defaults-file={config_file}" _START_SERVER_CMD_UNIX = ( u"export MYSQLD_PARENT_PID=$$\nexport MYSQLD_RESTART_EXIT=16\n\n" u"while true ; do\n {start_server_cmd} $*\n " u"if [ $? -ne $MYSQLD_RESTART_EXIT ]; then\n break\n fi\ndone" u"".format(start_server_cmd=_START_SERVER_CMD)) # Do not use/set MYSQLD_PARENT_PID=1 on Windows, otherwise the SHUTDOWN statement will not work properly, stopping # the wrong sandbox. # NOTE: Both RESTART and SHUTDOWN work properly on Windows without MYSQLD_PARENT_PID (not needed). _START_SERVER_CMD_WIN = ( u"chcp {code_page}\nset MYSQLD_RESTART_EXIT=16\n\n:while\n" u"{start_server_cmd} \nIF %ERRORLEVEL% EQU %MYSQLD_RESTART_EXIT% (\n" u" goto :while\n)\nEXIT /B %ERRORLEVEL%" u"".format(start_server_cmd=_START_SERVER_CMD, code_page=_CURRENT_ANSI_CODE_PAGE)) _STOP_SERVER_CMD = (u"{mysqladmin_path} --defaults-file={config_file} " u"shutdown -p") _CREATE_RSA_SSL_FILES_CMD = u"{mysql_ssl_rsa_setup_path} --datadir={datadir}" _WIN_SCRIPT = u"@echo off\necho {message}\n{content}\n" _UNIX_SCRIPT = u"#!/bin/bash\n\necho '{message}'\n{content}\n" # Timeout to wait for mysqld to start listening to connections SANDBOX_TIMEOUT = 30 _MAX_RMTREE_RETRIES = 5 DEFAULT_SANDBOX_DIR = "~/mysql-sandboxes" _LOCKFILE_NAME = "lockfile" _SERVER_READY_LOG_MESSAGES = ("mysqld: ready for connections.", "mysqld.exe: ready for connections.") # Sandbox commands SANDBOX = "sandbox" SANDBOX_START = "start" SANDBOX_STOP = "stop" SANDBOX_KILL = "kill" SANDBOX_CREATE = "create" SANDBOX_DELETE = "delete" # Error messages _ERROR_OVERRIDE_PORT = ("Overriding the port value is not supported. Please " "use the --port option to specify a different port " "when creating the sandbox instance.") _ERROR_CREATE_DIR = ("Unable to create {dir} directory '{dir_path}': " "{error}") _ERROR_NOT_CREATED = ("Cannot start MySQL sandbox for the given port because " "it does not exist.") _ERROR_VERSION_NOT_SUPPORTED = ("Provided mysqld executable '{0}' has a non " "supported version: '{1}'. MySQL version must " "be >= '{2}' and < '{3}'.") _ERROR_CANNOT_FIND_TOOL = ("Could not find {exec_name} executable. " "Make sure it is on the {path_var_name} " "environment variable.") _ERROR_CANNOT_FIND_VALID_TOOL = ( "Could not find a valid {exec_name} executable with a supported version " "(>= {min_ver} and < {max_ver}). Make sure it is on the {path_var_name} " "environment variable.") _ERROR_CHECK_VALID_TOOL = "Could not verify {exec_name} executable: {error}." def _create_start_script(script_name, script_path, mysqld, config_file): """Create a script file to start the Sandbox instance. :param script_name: Name of the script without the extension. :type script_name: str :param script_path: absolute path to the directory were the script will be created. :type script_path: str :param mysqld: absolute path to the mysqld executable. :type mysqld: str :param config_file: absolute path to the configuration file to be used by the sandbox. :type config_file: str :return: Path of the created script :rtype: str """ start_path = os.path.join(script_path, script_name) if os.name == "nt": script_contents = _WIN_SCRIPT.format( message="Starting MySQL sandbox", content=_START_SERVER_CMD_WIN) # add windows script extension start_path += ".bat" else: script_contents = _UNIX_SCRIPT.format( message="Starting MySQL sandbox", content=_START_SERVER_CMD_UNIX) # add unix script extension start_path += ".sh" script_contents = script_contents.format( mysqld_path=tools.shell_quote(os.path.normpath(mysqld)), config_file=tools.shell_quote(os.path.normpath(config_file))) _LOGGER.debug("Creating start script on '%s'", start_path) enc_start_path = tools.fs_encode(start_path) try: with open(enc_start_path, "w") as f: f.write(tools.fs_encode(script_contents)) except Exception as err: raise exceptions.GadgetError("Unable to create start script for " "sandbox instance", cause=err) _LOGGER.debug("Start script '%s' successfully created.", start_path) if os.name == "posix": # No need to set exec permissions on windows, also python can't do it. _LOGGER.debug("Changing permissions of start script to 700") os.chmod(enc_start_path, 0o700) _LOGGER.debug("Permissions changed successfully.") return start_path def _create_stop_script(script_name, script_path, mysqladmin, config_file): """Create a script file to stop the Sandbox instance. :param script_name: Name of the script without the extension. :type script_name: str :param script_path: absolute path to the directory were the script will be created. :type script_path: str :param mysqladmin: absolute path to the mysqladmin executable. :type mysqladmin: str :param config_file: absolute path to the configuration file to be used by the sandbox. :type config_file: str :return: Path of the created script :rtype: str """ stop_path = os.path.join(script_path, script_name) stop_msg = ("Stopping MySQL sandbox using mysqladmin shutdown... " "Root password is required.") if os.name == "nt": script_contents = _WIN_SCRIPT.format( message=stop_msg, content=_STOP_SERVER_CMD) # add windows script extension stop_path += ".bat" else: script_contents = _UNIX_SCRIPT.format( message=stop_msg, content=_STOP_SERVER_CMD) # add unix script extension stop_path += ".sh" _LOGGER.debug("Creating stop script on '%s'", stop_path) enc_stop_path = tools.fs_encode(stop_path) try: with open(enc_stop_path, "w") as f: f.write(tools.fs_encode(script_contents.format( mysqladmin_path=tools.shell_quote(mysqladmin), config_file=tools.shell_quote(config_file)))) except Exception as err: raise exceptions.GadgetError("Unable to create stop script for " "sandbox instance", cause=err) _LOGGER.debug("Stop script '%s' successfully created.", stop_path) if os.name == "posix": # No need to set exec permissions on windows, also python can't do it. _LOGGER.debug("Changing permissions of stop script to 700") os.chmod(enc_stop_path, 0o700) _LOGGER.debug("Permissions changed successfully.") return stop_path def _find_basedir(mysqld_path): """Try to find the basedir of a server using the path of the mysqld exec. :param mysqld_path: Path of the mysqld executable :type mysqld_path: str :return: the path of the basedir if we can find it :rtype: str :raises: GadgetError if unable to find the basedir """ _LOGGER.debug("Trying to find basedir for mysqld executable '%s'", mysqld_path) # get the real path, any symbolic links mysqld_path = os.path.realpath(mysqld_path) # base directory were we will start searching. Usually executables are in # a bin folder base = os.path.abspath(os.path.join(mysqld_path, os.pardir, os.pardir)) if os.path.isdir(base): _LOGGER.debug("Guessing basedir is at '%s'", base) return base else: raise exceptions.GadgetError("Could not find basedir for mysqld " "executable '{0}'".format(mysqld_path)) def _get_sandbox_dirs(**kwargs): """ Calculates the absolute sandbox paths from the kwargs dict. Returns the absolute path for the sandbox_base_dir and for the sandbox_dir from the kwargs dict provided. :param kwargs: Keyword arguments: sandbox_base_dir: base path for the created MySQL sandbox instances. Default is DEFAULT_SANDBOX_DIR. port: The port where the sandbox will listen for MySQL connections. :type kwargs: dict :raises GadgetError: If the port is not specified. :returns: A tuple with the absolute path for the sandbox_base_dir as the first element and the absolute path for the sandbox as the second element. :rtype: tuple """ # get mandatory values try: port = int(kwargs["port"]) except KeyError: raise exceptions.GadgetError("It is mandatory to specify a port.") # Get default values for optional variables sandbox_base_dir = os.path.normpath(os.path.expanduser(kwargs.get( "sandbox_base_dir", DEFAULT_SANDBOX_DIR))) # Convert sandbox_base_dir to an absolute path if it is not already one. sandbox_base_dir = tools.get_abs_path(sandbox_base_dir, os.getcwd()) return sandbox_base_dir, os.path.join(sandbox_base_dir, str(port)) def _set_secure_file_priv(opt_override_dict, sandbox_dir): """Verify and update the secure_file_priv value. This methods checks if the secure_file_priv has been given with the --opt option, and in such case it creates the directory if it does not exists. If the value for secure_file_priv is not a full path (only a folder name) then a folder with the given name will be created inside the sandbox directory. Otherwise, if no secure_file_priv is given with the --opt, by default the secure_file_priv is overwritten and set to the 'mysql-files' folder inside the sandbox directory. :param opt_override_dict: The options provided with the --opt option to add/update the configuration file. :type opt_override_dict: dict :param sandbox_dir: The path used by this sandbox (including the port). :type sandbox_dir: string :raise GadgetError: If the directory for the secure_file_priv can not be created. """ # Retrieve the secure_file_priv in case is provided. secure_file_priv = opt_override_dict.get('secure_file_priv', None) if secure_file_priv is None: # No value set for secure_file_priv, overwrite the default value. secure_file_priv = os.path.join(sandbox_dir, "mysql-files") enc_secure_file_priv = tools.fs_encode(secure_file_priv) # Check if secure_file_priv exists: if not os.path.isdir(enc_secure_file_priv): # Try to create it if it does not exist try: os.makedirs(enc_secure_file_priv) except OSError as err: raise exceptions.GadgetError( _ERROR_CREATE_DIR.format(dir="secure-file-priv", dir_path=secure_file_priv, error=str(err))) # Update dictionary of options to overwrite the configuration file. opt_override_dict['secure_file_priv'] = \ secure_file_priv.replace("\\", "/") else: # Check if the directory for secure_file_priv exists and create it # if need, otherwise nothing need to be done. secure_file_priv = os.path.normpath(os.path.expanduser( secure_file_priv)) if not os.path.isdir(secure_file_priv): # Try to create it if it does not exist try: # if not a full path, create it in sandbox_dir tail, head = os.path.split(secure_file_priv) if not tail: secure_file_priv = os.path.join(sandbox_dir, head) if not os.path.isdir(secure_file_priv): os.makedirs(secure_file_priv) except OSError as err: raise exceptions.GadgetError( _ERROR_CREATE_DIR.format(dir="secure-file-priv", dir_path=secure_file_priv, error=str(err))) # Update dictionary of options to overwrite the configuration file. opt_override_dict["secure_file_priv"] = \ secure_file_priv.replace("\\", "/") def sandbox_exists(**kwargs): """Checks if a MySQL sandbox already exists. Returns true in case it exists and False otherwise. :param kwargs: Keyword arguments: port: The port where the sandbox will listen for MySQL connections. sandbox_base_dir: base path for the created MySQL sandbox instances. Default is DEFAULT_SANDBOX_DIR. :type kwargs: dict :return: True if sandbox exists and False otherwise. :rtype: bool """ # Get sandbox paths _, sandbox_dir = _get_sandbox_dirs(**kwargs) enc_sandbox_dir = tools.fs_encode(sandbox_dir) return os.path.isdir(enc_sandbox_dir) and os.listdir(enc_sandbox_dir) # pylint: disable=R0915, R0914 def create_sandbox(**kwargs): """Create a new MySQL sandbox. :param kwargs: Keyword arguments: port: The port where the sandbox will listen for MySQL connections. passwd: password to be used for the root account in the MySQL sandbox. basedir: the directory that will be used as the basedir of the MySQL sandbox instance. mysqlx_port: the port where the sandbox will listen for the X-Protocol connections. Default value is <port>*10. sandbox_base_dir: base path for the created MySQL sandbox instances. Default is DEFAULT_SANDBOX_DIR. mysqld_path: Path to the mysqld executable. By default it will search the PATH of the system. mysqladmin_path: Path to the mysqladmin executable. By default it will search the PATH of the system. mysql_ssl_rsa_setup_path: Path to the mysql_ssl_rsa_setup executable. By default it will search the PATH of the system. server_id: Server-id value of the MySQL sandbox instance. By default a random id is used. opt: list of additional values to save under the [mysqld] section of the option file. timeout: timeout in seconds to wait for the sandbox instance to start listening for connections. ignore_ssl_error: If false (default) the sandbox must be created with support for SSL throwing an error if SSL support cannot be added. If true no error will be issued if SSL support cannot be provided and SSL support will be skipped. start: if true leave the sandbox running after its creation :type kwargs: dict """ # get mandatory values try: port = int(kwargs["port"]) except KeyError: raise exceptions.GadgetError("It is mandatory to specify a port.") password = kwargs.get("passwd") ignore_ssl_error = kwargs.get("ignore_ssl_error", False) start = kwargs.get("start", False) # Get default values for optional variables timeout = kwargs.get("timeout", SANDBOX_TIMEOUT) mysqlx_port = int(kwargs.get("mysqlx_port", port * 10)) # Verify if mysqlx port is valid. if (mysqlx_port < 1024 or mysqlx_port > 65535) and \ "mysqlx_port" not in kwargs.keys(): raise exceptions.GadgetError( "Invalid X port '{0}', it must be >= 1024 and <= 65535. " "Use a lower value for 'port' to generate a valid X port " "(by default, portx = port * 10), or use the 'portx' " "option to specify a custom value.".format(mysqlx_port)) _, sandbox_dir = _get_sandbox_dirs(**kwargs) enc_sandbox_dir = tools.fs_encode(sandbox_dir) # Check if sandbox_dir is empty if os.path.isdir(enc_sandbox_dir) and os.listdir(enc_sandbox_dir): raise exceptions.GadgetError(u"The sandbox dir '{0}' is not empty." u"".format(sandbox_dir)) # If no value is provided for mysqld, search value on PATH and default # mysqld paths. try: mysqld_path = kwargs.get("mysqld_path", tools.get_tool_path( None, "mysqld", search_path=True, required=True, check_tool_func=server.is_valid_mysqld)) except exceptions.GadgetError as err: if err.errno == 1: raise exceptions.GadgetError(_ERROR_CANNOT_FIND_TOOL.format( exec_name="mysqld", path_var_name=PATH_ENV_VAR)) elif err.errno == 2: raise exceptions.GadgetError(_ERROR_CANNOT_FIND_VALID_TOOL.format( exec_name="mysqld", min_ver='.'.join(str(i) for i in MIN_MYSQL_VERSION), max_ver='.'.join(str(i) for i in MAX_MYSQL_VERSION), path_var_name=PATH_ENV_VAR)) else: raise exceptions.GadgetError(_ERROR_CHECK_VALID_TOOL.format( exec_name="mysqld", error=err.errmsg)) # If no value is provided for mysqladmin, by default search value on PATH mysqladmin_path = kwargs.get("mysqladmin_path", tools.get_tool_path(None, "mysqladmin", search_path=True, required=False)) if not mysqladmin_path: raise exceptions.GadgetError(_ERROR_CANNOT_FIND_TOOL.format( exec_name="mysqladmin", path_var_name=PATH_ENV_VAR)) # If no value is provided for mysql_ssl_rsa_setup, by default search value # on PATH mysql_ssl_rsa_setup_path = kwargs.get( "mysql_ssl_rsa_setup_path", tools.get_tool_path( None, "mysql_ssl_rsa_setup", search_path=True, required=False)) if not mysql_ssl_rsa_setup_path and not ignore_ssl_error: raise exceptions.GadgetError( _ERROR_CANNOT_FIND_TOOL.format(exec_name="mysql_ssl_rsa_setup", path_var_name=PATH_ENV_VAR)) # Checking if mysql, mysqladmin and mysql_ssl_rsa_setup meet requirements if not tools.is_executable(mysqld_path): raise exceptions.GadgetError( "Provided mysqld '{0}' is not a valid executable." "".format(mysqld_path)) if not tools.is_executable(mysqladmin_path): raise exceptions.GadgetError( "Provided mysqladmin '{0}' is not a valid executable." "".format(mysqladmin_path)) if not ignore_ssl_error and not \ tools.is_executable(mysql_ssl_rsa_setup_path): raise exceptions.GadgetError( "Provided mysql_ssl_rsa_setup '{0}' is not a valid executable." "".format(mysql_ssl_rsa_setup_path)) mysqld_ver, version_str = server.get_mysqld_version(mysqld_path) if not MIN_MYSQL_VERSION <= mysqld_ver < MAX_MYSQL_VERSION: raise exceptions.GadgetError( _ERROR_VERSION_NOT_SUPPORTED.format( mysqld_path, version_str, '.'.join(str(i) for i in MIN_MYSQL_VERSION), '.'.join(str(i) for i in MAX_MYSQL_VERSION))) basedir = kwargs.get("basedir", None) if basedir is None: # If no value was provided, try to guess it from mysqld try: basedir = _find_basedir(mysqld_path) except exceptions.GadgetError: raise exceptions.GadgetError( "Unable to find the basedir for mysqld executable '{0}'. " "Please use the --basedir option to specify it." "".format(mysqld_path)) # By default a random ID value. server_id = kwargs.get("server_id", server.generate_server_id()) # Get list of options to override mysqld_opts = kwargs.get("opt", []) opt_override_dict = option_list_to_dictionary(mysqld_opts) # Datadir datadir = os.path.join(sandbox_dir, "sandboxdata") # Binary dir sandbox_bin_dir = os.path.join(sandbox_dir, "bin") enc_sandbox_bin_dir = tools.fs_encode(sandbox_bin_dir) # pid file_path pidf_path = os.path.join(sandbox_dir, "{0}.pid".format(port)) # Initialize new mysql sandbox # pylint: disable=E1101 _LOGGER.step("Initializing new MySQL sandbox on '%s'.", sandbox_dir) # Check if sandbox_dir exists: if not os.path.isdir(enc_sandbox_dir): # Try to create it if it does not exist try: os.makedirs(enc_sandbox_dir) os.makedirs(enc_sandbox_bin_dir) except OSError as err: raise exceptions.GadgetError( _ERROR_CREATE_DIR.format(dir="sandbox", dir_path=sandbox_dir, error=unicode(err))) # Update opt_override_dict with value used for secure_file_priv. _set_secure_file_priv(opt_override_dict, sandbox_dir) _LOGGER.debug("Option secure_file_priv will be set with value: %s", opt_override_dict['secure_file_priv']) # Create option file dictionary # MySQL prefers Unix stile paths, so convert all paths to unix style opt_dict = {"mysqld": { "port": port, "loose_mysqlx_port": mysqlx_port, "server_id": server_id, "socket": "mysqld.sock", "loose_mysqlx_socket": "mysqlx.sock", "basedir": basedir.replace("\\", "/"), "datadir": datadir.replace("\\", "/"), "report_port": port, "report_host": "127.0.0.1", "log_error": os.path.join(datadir, "error.log").replace("\\", "/"), "binlog_checksum": "NONE", "gtid_mode": "ON", "transaction_write_set_extraction": "XXHASH64", "binlog_format": "ROW", "log_bin": None, "enforce_gtid_consistency": "ON", "pid_file": pidf_path.replace("\\", '/'), }, "client": { "port": port, "user": "root", "protocol": "TCP", }} # master_info_repository and relay_log_info_repository were deprecated in # 8.0.23 and the setting TABLE is the default since then if mysqld_ver < (8, 0, 23): opt_dict["mysqld"]["master_info_repository"] = "TABLE" opt_dict["mysqld"]["relay_log_info_repository"] = "TABLE" # log_slave_updates is ON by default since 8.0.3 if mysqld_ver < (8, 0, 3): opt_dict["mysqld"]["log_slave_updates"] = "ON" if mysqld_ver < (8, 0, 13): # Disable syslog to avoid issue on Windows. opt_dict["mysqld"]["loose_log_syslog"] = "OFF" # Enable mysql_cache_cleaner plugin on server versions = 8.0.4. # This plugin is required for the hash based authentication to work # (caching_sha2_password) to allow the shell to connect using the X # protocol if SSL is disabled. if mysqld_ver == (8, 0, 4): opt_dict["mysqld"]["mysqlx_cache_cleaner"] = "ON" # Starting on mysql 8.0.21, group replication supports binlog_checksum # so we can remove the NONE requirement from opt dict an use the # server default (CRC32) if mysqld_ver >= (8, 0, 21): del opt_dict["mysqld"]["binlog_checksum"] # Starting with MySQL 8.0.23, having parallel-appliers enabled is a # requirement for InnoDB cluster/ReplicaSet usage. # So when deploying sandboxes, we already enable those settings if mysqld_ver >= (8, 0, 23): opt_dict["mysqld"]["binlog_transaction_dependency_tracking"] = "WRITESET" if mysqld_ver >= (8, 0, 26): opt_dict["mysqld"]["replica_preserve_commit_order"] = "ON" opt_dict["mysqld"]["replica_parallel_type"] = "LOGICAL_CLOCK" opt_dict["mysqld"]["replica_parallel_workers"] = 4 else: opt_dict["mysqld"]["slave_preserve_commit_order"] = "ON" opt_dict["mysqld"]["slave_parallel_type"] = "LOGICAL_CLOCK" opt_dict["mysqld"]["slave_parallel_workers"] = 4 # MySQLx plugin is automatically loaded starting from versions 8.0.11. if mysqld_ver < (8, 0, 11): opt_dict["mysqld"]["plugin_load"] = \ "mysqlx.so" if os.name == "posix" else "mysqlx.dll" if opt_override_dict: # If port is one of the options to override raise exception _LOGGER.debug("Adding/Overriding option file values.") if "port" in opt_override_dict: raise exceptions.GadgetError(_ERROR_OVERRIDE_PORT) # override mysqld dict with options received from cmd line opt_dict["mysqld"].update(opt_override_dict) # Create option file optf_path = create_option_file(opt_dict, "my.cnf", sandbox_dir) # If on Linux, create a temporary copy of the mysqld binary to avoid # possible AppArmor or SELinux issues. # Note: Creating a symbolic link will not solve the problem. if os.name != "nt" and sys.platform != "darwin": local_mysqld_path = os.path.join(sandbox_dir, "bin", "mysqld") try: _LOGGER.debug(u"Copying mysqld binary '%s' to '%s'", mysqld_path, local_mysqld_path) shutil.copy(tools.fs_encode(mysqld_path), tools.fs_encode(local_mysqld_path)) mysql_bindir = os.path.dirname(mysqld_path) # Symlink possibly bundled OpenSSL shared libs for name in os.listdir(tools.fs_encode(mysql_bindir)): if name.startswith("lib") and ".so" in name: path = os.path.join(mysql_bindir, name) new_path = os.path.join(sandbox_dir, "bin", name) _LOGGER.debug(u"Symlinking '%s' to '%s'", path, new_path) os.symlink(tools.fs_encode(path), tools.fs_encode(new_path)) except (IOError, shutil.Error) as err: raise exceptions.GadgetError( u"Unable to copy mysqld binary '{0}' to '{1}': '{2}'." u"".format(mysqld_path, sandbox_dir, unicode(err))) # Copies the protobuf libraries when they are bundled in the package # it is assumed that id not bundled they are system wide and the mysqld # binary will be able to find them. # TODO(rennox): This should be turned into a function like get_tool_path # As right now it is handling the cases of working with a package using # the variants used by PB2 on the tests, but it does not guarantee it # will work with a system installed MySQL. if mysqld_ver >= (8, 0, 18): library_path = os.path.join(basedir, "lib", "mysql", "private") sandbox_lib_dir = os.path.join(sandbox_dir, "lib", "mysql", "private") if not os.path.exists(library_path): library_path = os.path.join(basedir, "lib64", "mysql", "private") sandbox_lib_dir = os.path.join(sandbox_dir, "lib64", "mysql", "private") if not os.path.exists(library_path): library_path = os.path.join(basedir, "lib", "private") sandbox_lib_dir = os.path.join(sandbox_dir, "lib", "private") path = "" if os.path.exists(library_path): enc_sandbox_lib_dir = tools.fs_encode(sandbox_lib_dir) try: os.makedirs(enc_sandbox_lib_dir) except OSError as err: raise exceptions.GadgetError( _ERROR_CREATE_DIR.format(dir="protobuf library", dir_path=sandbox_lib_dir, error=str(err))) try: for name in os.listdir(tools.fs_encode(library_path)): if name.startswith("lib") and ".so" in name: path = os.path.join(library_path, name) target_path = os.path.join(sandbox_lib_dir, name) _LOGGER.debug(u"Copying library '%s' to '%s'", path, sandbox_lib_dir) shutil.copy(tools.fs_encode(path), tools.fs_encode(target_path)) except (IOError, shutil.Error) as err: raise exceptions.GadgetError( u"Unable to copy mysqld library '{0}' to '{1}': '{2}'." u"".format(path, sandbox_lib_dir, unicode(err))) else: local_mysqld_path = mysqld_path # Get the command string create_cmd = _CREATE_SANDBOX_CMD.format( mysqld_path=tools.shell_quote(local_mysqld_path), config_file=tools.shell_quote(os.path.normpath(optf_path))) # If we are running the script as root , the --user=root option is needed if os.name == "posix" and getpass.getuser() == "root": _LOGGER.warning("Creating a sandbox as root is not recommended.") create_cmd = "{0} --user=root".format(create_cmd) # Fake PID to avoid the server starting the monitoring process if os.name == "nt": os.environ['MYSQLD_PARENT_PID'] = "{0}".format(port) init_proc = tools.run_subprocess(create_cmd, shell=False, stderr=subprocess.PIPE) _, stderr = init_proc.communicate() if init_proc.returncode != 0: raise exceptions.GadgetError( f"Error initializing MySQL sandbox '{port}'. '{create_cmd}' failed\ with return code '{init_proc.returncode}' and message: {stderr.strip()}'.") _LOGGER.debug("Creating SSL/RSA files.") enc_datadir = tools.fs_encode(datadir) # Create SSL/RSA Files if they don't exist using the mysql_ssl_rsa_setup. if (os.path.isfile(os.path.join(enc_datadir, "ca.pem")) or os.path.isfile(os.path.join(enc_datadir, "server-cert.pem")) or os.path.isfile(os.path.join(enc_datadir, "server-key.pem"))): _LOGGER.debug("SSL/RSA files already exist.") elif mysql_ssl_rsa_setup_path: # MySQL servers have the capability of automatically generating # missing SSL and RSA files at startup, for MySQL distributions # compiled using OpenSSL. For MySQL distributions using YaSSL this can # be done manually using the mysql_ssl_rsa_setup utility however it # requires the openssl command to be available. rsa_ssl_cmd = _CREATE_RSA_SSL_FILES_CMD.format( mysql_ssl_rsa_setup_path=tools.shell_quote( mysql_ssl_rsa_setup_path), datadir=tools.shell_quote(datadir)) create_ssl_rsa_files_proc = tools.run_subprocess( rsa_ssl_cmd, shell=False, stderr=subprocess.PIPE) _, stderr = create_ssl_rsa_files_proc.communicate() # if the return code is not 0, an error occurred. Raise an exception # and show it if the ignore_ssl_error flag was not used. if create_ssl_rsa_files_proc.returncode: message = ( "Unable to create SSL/RSA files. mysql_ssl_rsa_setup exited " "with error code '{0}' and message: '{1}'.".format( create_ssl_rsa_files_proc.returncode, stderr.strip())) if ignore_ssl_error: _LOGGER.warning(message) else: raise exceptions.GadgetError( "{0}. You can use the option to ignore SSL errors to skip " "the use of SSL.".format(message)) else: # if the process ran without any errors _LOGGER.debug("SSL/RSA files created.") else: # No SSL/RSA files created if mysql_ssl_rsa_setup is not available. # NOTE: Error already raised previously if the tool is not found and # ignore_ssl_error is False. _LOGGER.debug("No SSL/RSA files created.") # create start script _LOGGER.debug("Creating start script for sandbox.") start_path = _create_start_script("start", sandbox_dir, local_mysqld_path, optf_path) _LOGGER.debug("Start script created.") # Create stop script _LOGGER.debug("Creating stop script for sandbox.") stop_path = _create_stop_script("stop", sandbox_dir, mysqladmin_path, optf_path) _LOGGER.debug("Stop script created.") # if the start option was provided but there is no need to change the # password. if start and not password: _LOGGER.info("Starting sandbox") if tools.is_listening("localhost", port): # there is already something running on the port that the sandbox # will use raise exceptions.GadgetError( "Unable to start the MySQL sandbox. Port '{0}' on which " "the sandbox runs was already in use.".format(port)) start_cmd = tools.shell_quote(start_path) if os.name == "posix" and getpass.getuser() == "root": # If we are running the script as root , the --user=root option is # needed start_cmd = "{0} {1}".format(start_cmd, tools.shell_quote("--user=root")) _LOGGER.debug("Launching mysqld") with open(os.devnull, "r") as devnull_in: with open(os.devnull, "w") as devnull_out: server_proc = tools.run_subprocess( start_cmd, stdin=devnull_in, stdout=devnull_out, stderr=devnull_out, close_fds=True) # wait until server is listening on the given port i = 0 _LOGGER.debug("Waiting for MySQL sandbox to start listening " "for connections on port '%i'", port) while i < timeout: if not tools.is_listening("localhost", port): time.sleep(1) i += 1 else: # port is listening, break out of loop _LOGGER.debug("MySQL sandbox is listening for " "connections on port '%i'", port) break else: # timeout occurred, send signal to terminate process try: server_proc.terminate() except Exception as err: raise exceptions.GadgetError( "Timeout waiting for mysqld process with pid '{0}'" " to start and we got error '{1} while trying to " "terminate it. You might need to terminate it " "manually.".format(server_proc.pid, str(err))) else: # server was successfully terminated raise exceptions.GadgetError( "Timeout waiting for mysqld process with pid " "'{0}' to start.".format(server_proc.pid)) # Change root password if one was provided # start the server if password: _LOGGER.info("Changing root password") if tools.is_listening("localhost", port): # there is already something running on the port that the sandbox # will use as such we cannot spawn the server to change the root # password. # lets remove the sandbox dir try: _LOGGER.debug("removing MySQL sandbox dir '%s'.", sandbox_dir) shutil.rmtree(sandbox_dir) finally: raise exceptions.GadgetError( "Unable to start the MySQL sandbox in order to change the " "root password. Port '{0}' on which the sandbox runs was " "already in use.".format(port)) start_cmd = tools.shell_quote(start_path) if os.name == "posix" and getpass.getuser() == "root": # If we are running the script as root , the --user=root option is # needed start_cmd = "{0} {1}".format(start_cmd, tools.shell_quote("--user=root")) _LOGGER.debug("Launching mysqld to change the root password") with open(os.devnull, "r") as devnull_in: with open(os.devnull, "w") as devnull_out: server_proc = tools.run_subprocess( start_cmd, stdin=devnull_in, stdout=devnull_out, stderr=devnull_out, close_fds=True) # wait until server is listening on the given port i = 0 _LOGGER.debug("Waiting for MySQL sandbox to start listening " "for connections on port '%i'", port) while i < timeout: if not tools.is_listening("localhost", port): time.sleep(1) i += 1 else: # port is listening, break out of loop _LOGGER.debug("MySQL sandbox is listening for " "connections on port '%i'", port) break else: # timeout occurred, send signal to terminate process try: server_proc.terminate() except Exception as err: raise exceptions.GadgetError( "Timeout waiting for mysqld process with pid '{0}'" " used to change root password to start and we got" " error '{1} while trying to terminate it. You " "might need to terminate it manually." "".format(server_proc.pid, str(err)) ) else: # server was successfully terminated raise exceptions.GadgetError( "Timeout waiting for mysqld process with pid '{0}'" " used to change root password to start." "".format(server_proc.pid) ) # server is now listening, connect to it and change root password s = server.Server({"conn_info": "root@localhost:{0}".format(port)}) try: s.connect() except exceptions.GadgetServerError as err: # Check if server process has finished start_ret_code = server_proc.poll() if start_ret_code is None: # server_proc is still running raise exceptions.GadgetError( "Cannot change root password, unable to connect to " "sandbox server: {0}".format(str(err))) else: # server_proc has ended unexpectedly, some error occurred raise exceptions.GadgetError( "Cannot change root password. Server stopped unexpectedly " "with return code '{0}'. Check error log file '{1}'." "".format(start_ret_code, os.path.join(datadir, "error.log"))) # change password of root account query = server.Query( "ALTER USER 'root'@'localhost' IDENTIFIED BY ?", server.Secret(password)) s.toggle_binlog(action="disable") s.exec_query(query) s.toggle_binlog(action="enable") s.disconnect() _LOGGER.info("Password changed.") if not start: _LOGGER.debug("Stopping mysqld.") try: # terminate server proc server_proc.terminate() except Exception as err: raise exceptions.GadgetError( "Unable to terminate the mysqld process with " "pid '{0}' that was started to change the root password: " "'{1}'. You might need to terminate it manually.".format( server_proc.pid, str(err))) server_proc.wait() _LOGGER.debug("mysqld stopped.") def start_sandbox(**kwargs): """Starts a MySQL sandbox. Start the server instance of an existing MySQL sandbox. Note: the sandbox must be created first, otherwise an error is issued. :param kwargs: Keyword arguments: port: The port where the sandbox will listen for MySQL connections. sandbox_base_dir: base path for the created MySQL sandbox instances. Default is DEFAULT_SANDBOX_DIR. mysqld_path: Path to the mysqld executable. By default it will search the PATH of the system. opt: list of additional values to save under the [mysqld] section of the option file. timeout: timeout in seconds to wait for the sandbox instance to start listening for connections. :type kwargs: dict :raises GadgetError: If the MySQL sandbox to start does not exist, for the given port and sandbox_base_dir. """ # If sandbox dir does not exist (not created) then raise an error. if not sandbox_exists(**kwargs): raise exceptions.GadgetError(_ERROR_NOT_CREATED) # Get mandatory values try: port = int(kwargs["port"]) except KeyError: raise exceptions.GadgetError("It is mandatory to specify a port.") # Get default values for optional variables timeout = int(kwargs.get("timeout", SANDBOX_TIMEOUT)) # Get list of options to override mysqld_opts = kwargs.get("opt", []) opt_override_dict = option_list_to_dictionary(mysqld_opts) _, sandbox_dir = _get_sandbox_dirs(**kwargs) # option_file path optf_path = os.path.join(sandbox_dir, "my.cnf") # Update opt_override_dict with value for secure_file_priv. _set_secure_file_priv(opt_override_dict, sandbox_dir) _LOGGER.debug("Option secure_file_priv will be set with value: %s", opt_override_dict['secure_file_priv']) # If sandbox already existed but option values were provided manually # add them to the option file if opt_override_dict: _LOGGER.debug("Adding/Overriding option file values.") # If port is one of the options to override raise exception if "port" in opt_override_dict: raise exceptions.GadgetError(_ERROR_OVERRIDE_PORT) mysql_opt_parser = MySQLOptionsParser(optf_path) for opt, val in opt_override_dict.items(): mysql_opt_parser.set("mysqld", opt, val) mysql_opt_parser.write() # Initialize mysql sandbox # pylint: disable=E1101 _LOGGER.step("Initializing MySQL sandbox on '%s'.", sandbox_dir) _LOGGER.info("Starting MySQL sandbox on port '%i'", port) # since the script starts mysqld server on background there is no simple # way to know if it started correctly, we will at least try to make sure # the port is not in use. if tools.is_listening("localhost", port): raise exceptions.GadgetError( "Unable to start MySQL sandbox because port '{0}' is already in " "use.".format(port)) # also check if mysqlx_port is in use if mysql_opt_parser.has_option("mysqld", "mysqlx_port"): # First we check for option without the loose prefix. If it exists # it means it was manually overridden by the user and it has precedence # over the loose-prefix one. mysqlx_port = mysql_opt_parser.get("mysqld", "mysqlx_port") elif mysql_opt_parser.has_option("mysqld", "loose_mysqlx_port"): mysqlx_port = mysql_opt_parser.get("mysqld", "loose_mysqlx_port") else: # if no mysqlx_port was specified, then there is no need to do any # validation. mysqlx_port = None if mysqlx_port is not None: if tools.is_listening("localhost", int(mysqlx_port)): raise exceptions.GadgetError( "Unable to start MySQL sandbox because port '{0}' for the X" " protocol is already in use.".format(mysqlx_port)) # Try to create lock file. If we're unable to do it, it means another # sandbox process is already running, so we raise an exception. flags = os.O_CREAT | os.O_EXCL | os.O_RDONLY lock_file_path = os.path.normpath(os.path.join(sandbox_dir, _LOCKFILE_NAME)) enc_lock_file_path = tools.fs_encode(lock_file_path) try: file_handle = os.open(enc_lock_file_path, flags) os.close(file_handle) # Create the start script since testutils sandbox operations copy over # the start script from the boilerplate directory and that will have the # path to the mysqld binary hardcoded (so we need to update the script # so it points to a correct mysqld binary). start_path = os.path.join(sandbox_dir, "start") if os.name == "nt": start_path += ".bat" else: start_path += ".sh" start_cmd = tools.shell_quote(start_path) if os.name == "posix" and getpass.getuser() == "root": _LOGGER.warning( "Starting a sandbox as root is not recommended.") # If we are running the script as root , the --user=root option is # needed start_cmd = "{0} {1}".format(start_cmd, tools.shell_quote("--user=root")) error_log_path = os.path.normpath( mysql_opt_parser.get("mysqld", "log_error")) enc_error_log_path = tools.fs_encode(error_log_path) if os.path.isfile(enc_error_log_path): # Find out last position of error log, before starting the process # since the error log persists several sessions. error_log_end_pos = os.path.getsize(enc_error_log_path) error_log_size = os.stat(enc_error_log_path).st_size else: # if error_log didn't exist, start reading at the beginning of the # file error_log_end_pos = 0 error_log_size = 0 with open(os.devnull, "r") as devnull_in: with open(os.devnull, "w") as devnull_out: server_proc = tools.run_subprocess( start_cmd, stdin=devnull_in, stdout=devnull_out, stderr=devnull_out, close_fds=True) started_at = time.time() started_ok = False _LOGGER.debug("Waiting for MySQL sandbox to start listening for " "connections on port '%i'", port) # Wait for the log file to be created by the server # in case it does not exist. i = 0 while i < timeout: if not os.path.isfile(enc_error_log_path): # Wait for the log file to be created by the server. time.sleep(1) i += 1 elif os.stat(enc_error_log_path).st_size > error_log_size: # Log file created or updated by the server and available. break else: # Wait for something to be written by the server time.sleep(1) i += 1 else: raise exceptions.GadgetError( u"Timeout waiting for the MySQL log error file to be " u"available: '{0}'. Please check configuration for " u"'log_error' in '{1}' file and that the log file is " u"accessible.".format(error_log_path, optf_path)) with open(enc_error_log_path, 'r') as f: # jump to current session position f.seek(error_log_end_pos) # Check that server has started correctly: while time.time() - started_at < timeout: # save last read position (start of line) last_pos = f.tell() line = f.readline() if not line: if not started_ok: # if nothing to read, wait a bit time.sleep(1) continue else: # Started ok and nothing else to read, exit loop. break if "\n" not in line: # if we didn't read a whole line, go back to saved position f.seek(last_pos) continue if not started_ok: # if we didn't find the ready_message yet, keep looking # for it. for server_ready_msg in _SERVER_READY_LOG_MESSAGES: if server_ready_msg in line: _LOGGER.debug( "MySQL sandbox is listening for " "connections on port '%i'", port) started_ok = True break if "[ERROR] Aborting\n" in line: # critical error, server won't start but will shutdown # on its own. raise exceptions.GadgetError( u"Unable to start server on port '{0}'. For more " u"information, check error log '{1}'".format( port, error_log_path)) if "[ERROR]" in line: _LOGGER.warning( "Error found during server startup: " "'%s'", line.strip()) if not started_ok: # timeout occurred, send signal to terminate process try: server_proc.terminate() except Exception as err: raise exceptions.GadgetError( "Timeout waiting for sandbox mysqld process with " "pid '{0}' to start and we got error '{1}' while " "trying to terminate it. You might need to " "terminate it manually." "".format(server_proc.pid, str(err))) else: # server was successfully terminated raise exceptions.GadgetError( u"Timeout waiting for sandbox mysqld process with " u"pid '{0}' to start. For more information, check " u"error log '{1}'.".format(server_proc.pid, error_log_path)) _LOGGER.info("MySQL sandbox running on port '%i' with process ID: " "'%i'", port, server_proc.pid) except OSError as err: if err.errno == errno.EEXIST: # Failed as the file already exists. raise exceptions.GadgetError( "Unable to lock sandbox directory. Another sandbox " "must be using it.") else: # Something unexpected went wrong so re-raise the exception. raise exceptions.GadgetError("Unexpected error: {0}".format( str(err))) finally: try: _LOGGER.debug(u"Removing lock file '%s'", lock_file_path) os.remove(enc_lock_file_path) _LOGGER.debug("Lock file removed") except OSError as err: if err.errno == errno.ENOENT: # file does not exist, ignore error pass else: raise exceptions.GadgetError( u"Unable to remove lock file '{0}': {1}".format( lock_file_path, unicode(err))) def stop_sandbox(**kwargs): """Stop an existing MySQL sandbox. :param kwargs: Keyword arguments: port: The port where the sandbox is listening for MySQL connections. sandbox_base_dir: base path for the created MySQL sandbox instances. Default is DEFAULT_SANDBOX_DIR. :type kwargs: dict """ # get mandatory values try: port = int(kwargs["port"]) except KeyError: raise exceptions.GadgetError("It is mandatory to specify a port.") password = kwargs.get("passwd") # Get default values for optional variables timeout = kwargs.get("timeout", SANDBOX_TIMEOUT) _, sandbox_dir = _get_sandbox_dirs(**kwargs) # Check if the pid file exists and if the port on which it is running is # still listening, meaning the sandbox is running. # pylint: disable=E1101 _LOGGER.step("Stopping MySQL sandbox on '%s'.", sandbox_dir) _LOGGER.debug("Executing SHUTDOWN SQL command localhost:%i", port) # Send shutdown signal conn_dict = {"user": "root", "host": "localhost", "port": port, "passwd": password} s = server.Server({"conn_info": conn_dict}) try: s.connect() except exceptions.GadgetServerError as err: if err.errno == 2026: # SSL error conn_dict["ssl"] = False s = server.Server({"conn_info": conn_dict}) try: s.connect() except exceptions.GadgetServerError as err: raise exceptions.GadgetError( "Unable to connect to MySQL sandbox {0} to send the " "SHUTDOWN request: '{1}'".format(str(s), str(err))) else: raise exceptions.GadgetError( "Unable to connect to MySQL sandbox {0} to send the " "SHUTDOWN request: '{1}'".format(str(s), str(err))) try: s.exec_query("SHUTDOWN") except exceptions.GadgetQueryError: # ignore query timeout or connection lost errors. pass # Wait for server to stop (listening on port). i = 0 _LOGGER.debug("Waiting for MySQL sandbox on port '%i' to stop.", port) while i < timeout: if tools.is_listening("localhost", port): time.sleep(1) i += 1 else: # port is listening, break out of loop _LOGGER.debug( "MySQL sandbox on port '%i' stopped.", port) break else: # Timeout occurred, issue an error raise exceptions.GadgetError( "Timeout waiting for sandbox at localhost:{0} to stop. " "You might need to terminate it manually " "or use the '{1} {2}' command.".format(port, SANDBOX, SANDBOX_KILL)) # Wait for the pid file to be deleted i = 0 pidf_path = os.path.join(sandbox_dir, "{0}.pid".format(port)) enc_pidf_path = tools.fs_encode(pidf_path) _LOGGER.debug("Waiting for MySQL Server pid file '%s' to de deleted.", pidf_path) while i < timeout: if os.path.exists(enc_pidf_path): time.sleep(1) i += 1 else: # pid was deleted, break out of loop _LOGGER.debug( "MySQL Server pid file '%s' deleted.", pidf_path) break else: # Timeout occurred, issue an error raise exceptions.GadgetError( "Timeout waiting for sandbox at localhost:{0} to stop. " "You might need to terminate it manually " "or use the '{1} {2}' command.".format(port, SANDBOX, SANDBOX_KILL)) # Server was stopped _LOGGER.info("MySQL sandbox was stopped on port '%i'.", port) def kill_sandbox(**kwargs): """Stop an existing MySQL sandbox. :param kwargs: Keyword arguments: port: The port where the sandbox is listening for MySQL connections. sandbox_base_dir: base path for the created MySQL sandbox instances. Default is DEFAULT_SANDBOX_DIR. :type kwargs: dict """ # get mandatory values try: port = int(kwargs["port"]) except KeyError: raise exceptions.GadgetError("It is mandatory to specify a port.") # Get default values for optional variables _, sandbox_dir = _get_sandbox_dirs(**kwargs) # pif file path pidf_path = os.path.join(sandbox_dir, "{0}.pid".format(port)) enc_pidf_path = tools.fs_encode(pidf_path) # Check if the pid file exists and if the port on which it is running is # still listening, meaning the sandbox is running. # pylint: disable=E1101 _LOGGER.step("Killing MySQL sandbox on '%s'.", sandbox_dir) # if a pid file still exists if os.path.exists(enc_pidf_path): _LOGGER.debug("Found pid file '%s'", pidf_path) # and a server listening on the port we specified if tools.is_listening("localhost", port): _LOGGER.debug("Found server listening on port '%i'", port) with open(enc_pidf_path) as f: pid = int(f.readline().strip()) _LOGGER.debug("Got pid '%i' from pid file '%s'", pid, pidf_path) _LOGGER.debug("Sending kill signal to process '%i'", pid) # kill process tools.stop_process_with_pid(pid, force=True) # remove pid file try: _LOGGER.debug("Removing pid file '%s'", pidf_path) os.unlink(enc_pidf_path) except OSError as err: _LOGGER.warning("Unable to remove pid file: '%s'", str(err)) else: # there is a process listening on the specified port, but there is # no pid file so emmit a warning and do nothing _LOGGER.warning("There is no MySQL sandbox listening on port %i, " "but a pid file was still found. Removing it.", port) try: os.unlink(enc_pidf_path) except OSError as err: _LOGGER.warning("Unable to remove pid file: '%s'", str(err)) else: # no pid file was found raise exceptions.GadgetError("Unable to find pid file. Kill " "operation will not proceed.") def delete_sandbox(**kwargs): """Deletes the folder of a MySQL sandbox. :param kwargs: Keyword arguments: port: The port where the sandbox is listening for MySQL connections. sandbox_base_dir: base path for the created MySQL sandbox instances. Default is DEFAULT_SANDBOX_DIR. :type kwargs: dict """ def on_delete_sandbox_error(_, path, exc_info): """Callback function to ignore the file not found errors for rmtree. """ # It will ignore file not found errors on delete operations type_, value, traceback = exc_info # if it is not a non existing file/folder (errno= 2) re raise exception if value.errno != 2: if value is not None: exc = type_(value) else: exc = type_ if sys.version_info[0] == 3: if exc.__traceback__ is not traceback: raise exc.with_traceback(traceback) raise exc else: # Log ignored exception raised when attempting to delete # non existing file/folder _LOGGER.debug("Ignored exception raised when trying to " "delete non-existing file/folder: '%s'", path) # get mandatory values try: port = int(kwargs["port"]) except KeyError: raise exceptions.GadgetError("It is mandatory to specify a port.") # Get default values for optional variables _, sandbox_dir = _get_sandbox_dirs(**kwargs) # pif file path pidf_path = os.path.join(sandbox_dir, "{0}.pid".format(port)) # Check if the pid file exists, meaning the sandbox is running and if the # port on which it is running is still listening. # pylint: disable=E1101 _LOGGER.step("Deleting MySQL sandbox on '%s'.", sandbox_dir) # if a pid file still exists if os.path.exists(tools.fs_encode(pidf_path)): _LOGGER.debug("Found pid file '%s'", pidf_path) # and a server listening on the port we specified, we cannot destroy it if tools.is_listening("localhost", port): _LOGGER.debug("Found server listening on port '%i'", port) raise exceptions.GadgetError( "Unable to delete sandbox folder: the MySQL sandbox instance " "on port '{0}' is running, please stop it to be able to " "delete it.".format(port)) else: _LOGGER.warning("A pid file was found but there is no MySQL " "sandbox listening on port '%i'. Sandbox will " "still be deleted.", port) # When deleting the sandbox, some files might still be in use as # the server might be shutting down. To account for this, we try # several times to delete the sandbox dir with increasing timeout # times. If we still fails, an exception is thrown. err = None for i in range(1, _MAX_RMTREE_RETRIES + 1): try: shutil.rmtree(sandbox_dir, onerror=on_delete_sandbox_error) break except OSError as err: _LOGGER.warning("Unable to delete MySQL sandbox folder " "'%s'. Retrying after '%d' seconds. '%d' " "retries left.", sandbox_dir, i, _MAX_RMTREE_RETRIES - i) time.sleep(i) else: # Failed to successfully remove the sandbox folder. raise exceptions.GadgetError( u"Unable to delete MySQL sandbox folder '{0}': '{1}'" u"".format(sandbox_dir, unicode(err))) else: # no pid file was found so we can safely delete try: shutil.rmtree(tools.fs_encode(sandbox_dir), onerror=on_delete_sandbox_error) except Exception as err: raise exceptions.GadgetError( u"Unable to delete MySQL sandbox folder '{0}': '{1}'" u"".format(sandbox_dir, unicode(err)), cause=err) mysql_gadgets/command/__init__.py 0000644 00000002241 15231164450 0013136 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # mysql_gadgets/__init__.py 0000644 00000024357 15231164450 0011534 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """mysql gadgets library""" import inspect import os import sys from mysql_gadgets.exceptions import GadgetError, GadgetVersionError # Major, Minor, Patch, Status VERSION = (2, 1, 0, '', 0) # Future versions will have to include only the X, Y (no Z). VERSION_STRING = "%s.%s.%s" % VERSION[0:3] COPYRIGHT_YEAR = "2016, 2020, " COPYRIGHT_FULL = ( "Copyright (c) %s Oracle and/or its affiliates.\n" "Oracle is a registered trademark of Oracle Corporation and/or its " "affiliates.\n" "Other names may be trademarks of their respective owners." % COPYRIGHT_YEAR) LICENSE = "GPLv2" VERSION_FRM = "{program} version %s" % VERSION_STRING LICENSE_FRM = (VERSION_FRM + "\n" + COPYRIGHT_FULL) PYTHON_MIN_VERSION = (2, 6, 0) PYTHON_MAX_VERSION = (4, 0, 0) # Minimum connector-python version that supports secure connection by default. CONNECTOR_MIN_VERSION = (2, 1, 7) # Using mathematical notation for intervals, supported MYSQL versions are as # follows: [MIN_MYSQL_VERSION, MAX_MYSQL_VERSION [ MIN_MYSQL_VERSION = (5, 7, 17) # minimum required version (supported) MAX_MYSQL_VERSION = (9, ) # maximum mysql version (not supported) MIN_PERSIST_MYSQL_VERSION = (8, 0, 11) # minimum mysql version that supports # set persist syntax. def check_expected_version(expected_version): """ Check the expected version of the used tools. Compare the given version with the current version of the modules/tools raising an exception if it is not compatible. For the expected version to be considered compatible with the current version, the major version number must be the same and the minor version number of the current version must be greater or equal than the expected version. The patch version number is ignored. Rationale: The major version number shall always be incremented if an incompatible change is made. The minor version number shall be incremented if new features are added in a backward-compatible manner, meaning that previous features are expected to continue to work (but the new added feature is only expected to be available starting from that minor version). The patch version is incremented for backward-compatible changes (e.g. bug fixes). :param expected_version: Excepted version to compare to the current one. :type expected_version: String :raises GadgetError: If the specified expected version value/format is invalid. :raises GadgetVersionError: If the expected version is not compatible with the current one. """ # Validate expected_version value and format. if expected_version: version_values = expected_version.split('.') if len(version_values) > 3: raise GadgetError("Invalid expected version value: '{0}'. Please " "specify at most 3 version number parts " "(format: MAJOR[.MINOR[.PATCH]])." "".format(expected_version)) else: raise GadgetError("Invalid expected version value: '{0}'. Please " "specify a valid version string (format: " "MAJOR[.MINOR[.PATCH]]).".format(expected_version)) # Get major, minor and patch version numbers. version_num = [-1, -1, -1] for idx, v_num in enumerate(version_values): if idx == 0: v_num_type = 'major' elif idx == 1: v_num_type = 'minor' else: v_num_type = 'patch' try: version_num[idx] = int(v_num) if version_num[idx] < 0: raise GadgetError("Invalid integer for the expected {0} " "version, it cannot be a negative number: " "'{1}'.".format(v_num_type, v_num)) except ValueError: raise GadgetError("Invalid integer for the expected {0} version " "number: '{1}'.".format(v_num_type, v_num)) # Validate the expected version comparing it to the current version. if (version_num[0] != VERSION[0]) or (version_num[1] > VERSION[1]): # The major expected version number must be the same as the one of the # current tool version. # The minor expected version number must be lower or equal than the one # of the current tool version. # The patch expected version number is ignored since it is associated # to compatible changes. raise GadgetVersionError( "The expected version is not compatible with the current version. " "Current version '{0}' and expected version: '{1}'." "".format('.'.join(map(str, VERSION[0:3])), expected_version)) def check_python_version(min_version=PYTHON_MIN_VERSION, max_version=PYTHON_MAX_VERSION, raise_exception_on_fail=False, name=None, exit_on_fail=True, return_error_msg=False, json_msg=False): """Check the Python version compatibility. By default this method uses constants to define the minimum and maximum Python versions required. It's possible to override this by passing new values on ``min_version`` and ``max_version`` parameters. It will run a ``sys.exit`` or raise a ``GadgetError`` if the version of Python detected it not compatible. :param min_version: Tuple with the minimum Python version required (inclusive). :type min_version: tuple :param max_version: Tuple with the maximum Python version required (exclusive). :type max_version: tuple :param raise_exception_on_fail: It will raise a ``GadgetError`` if True and Python detected is not compatible. :type raise_exception_on_fail: boolean :param name: Custom name for the script, if not provided it will get the module name from where this function was called. :type name: string :param exit_on_fail: If True, issue exit() else do not exit() on failure. :type exit_on_fail: boolean :param return_error_msg: If True, and is not compatible returns (result, error_msg) tuple. :type return_error_msg: boolean :param json_msg: If true return the error message in JSON format '{"type": "ERROR", "msg": "..."}' when exit_on_fail=True. By default: False, JSON format not used. :type json_msg: boolean :return: True if Python version is compatible, otherwise False if 'raise_exception_on_fail' is set to False or a tuple (False, error message) if 'return_error_msg' is set to True. :rtype: boolean or tuple :raises GadgetVersionError: if the detected Python version is not compatible and 'raise_exception_on_fail' is set to True. """ # Only use the fields: major, minor and micro sys_version = sys.version_info[:3] # Test min version compatibility is_compat = min_version <= sys_version # Test max version compatibility if it's defined if is_compat and max_version: is_compat = sys_version < max_version if not is_compat: if not name: # Get the utility name by finding the module # name from where this function was called frm = inspect.stack()[1] mod = inspect.getmodule(frm[0]) mod_name = os.path.splitext( os.path.basename(mod.__file__))[0] name = '%s gadget' % mod_name # Build the error message if max_version: max_version_error_msg = 'or higher and lower than %s' % \ '.'.join([str(el) for el in max_version]) else: max_version_error_msg = 'or higher' error_msg = ( 'Python version %(min_version)s %(max_version_error_msg)s is ' 'required. The version of Python detected was ' '%(sys_version)s. You may need to install or redirect the ' 'execution to an environment that includes a compatible Python ' 'version.' ) % { 'sys_version': '.'.join([str(el) for el in sys_version]), 'min_version': '.'.join([str(el) for el in min_version]), 'max_version_error_msg': max_version_error_msg } if raise_exception_on_fail: raise GadgetVersionError(error_msg) if exit_on_fail: if json_msg: sys.exit('{"type": "ERROR", "msg": "%s"}\n' % error_msg) else: sys.exit("ERROR: %s\n" % error_msg) if return_error_msg: return is_compat, error_msg return is_compat mysql_gadgets/exceptions.py 0000644 00000011452 15231164450 0012146 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """ This file contains the exceptions used by Gadgets and their libraries. """ class GadgetError(Exception): """Base class for Gadgets. This exception class is used to report errors gadgets and its libraries (modules), being used to communicate expected errors to the user. """ def __init__(self, message, errno=0, cause=None): super(GadgetError, self).__init__() self.args = (message, errno, cause) self.errmsg = message self.errno = errno self.cause = cause def __str__(self): return self.errmsg class GadgetLogError(GadgetError): """Errors for the logger module. Note: No logging feature might be available yet at the time this kind of error is raised. """ def __init__(self, message, errno=0, cause=None): super(GadgetLogError, self).__init__(message, errno, cause) class GadgetCnxInfoError(GadgetError): """Error when the server connection information is not valid. """ def __init__(self, message, errno=0, cause=None): super(GadgetCnxInfoError, self).__init__(message, errno, cause) class GadgetCnxFormatError(GadgetError): """Error parsing connection information (wrong format). """ def __init__(self, message, errno=0, cause=None): super(GadgetCnxFormatError, self).__init__(message, errno, cause) class GadgetServerError(GadgetError): """Error with the server. """ def __init__(self, message, errno=0, cause=None, server=None): super(GadgetServerError, self).__init__(message, errno, cause) self.server = server self.args = (message, errno, cause, server) def __str__(self): if self.server: return '{0} - {1}'.format(self.server, self.errmsg) else: return self.errmsg class GadgetCnxError(GadgetServerError): """Error with the server connection. Note: This is a subclass of the GadgetServerError. """ def __init__(self, message, errno=0, cause=None, server=None): super(GadgetCnxError, self).__init__(message, errno, cause, server) class GadgetQueryError(GadgetServerError): """Error when executing a statement (query) on the server. Note: This is a subclass of the GadgetServerError. """ def __init__(self, message, query, errno=0, cause=None, server=None): super(GadgetQueryError, self).__init__(message, errno, cause, server) self.query = query self.args = (message, query, errno, cause, server) def __str__(self): return '{0}. Query: {1}'.format( super(GadgetQueryError, self).__str__(), self.query) class GadgetDBError(GadgetServerError): """Error when a database operations fail. Note: This is a subclass of the GadgetServerError. """ def __init__(self, message, errno=0, cause=None, server=None, db=None): super(GadgetDBError, self).__init__(message, errno, cause, server) self.db = db self.args = (message, errno, cause, server, db) class GadgetRPLError(GadgetError): """Error during replication operations. """ def __init__(self, message, errno=0, cause=None): super(GadgetRPLError, self).__init__(message, errno, cause) class GadgetConfigParserError(GadgetError): """Error during handling of option files (config parser). """ def __init__(self, message, errno=0, cause=None): super(GadgetConfigParserError, self).__init__(message, errno, cause) class GadgetVersionError(GadgetError): """Error when checking the version of a component. """ def __init__(self, message, errno=0, cause=None): super(GadgetVersionError, self).__init__(message, errno, cause) mysql_gadgets/__main__.py 0000644 00000015751 15231164450 0011513 0 ustar 00 #!/usr/bin/env python # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """ This script is used for manage MySQL Group Replication (GR), providing the commands to check a server configuration compliance with GR requirements, start a replica set, add and remove servers from a group, cloning and displaying the health status of the members of the GR group. It also allows the creation of MySQL sandbox instances. """ # pylint: disable=wrong-import-position,wrong-import-order import io import logging import os import signal import sys import json from mysql_gadgets.common.logger import setup_logging, CustomLevelLogger from mysql_gadgets.common.tools import fs_decode from mysql_gadgets.command.sandbox import create_sandbox, stop_sandbox, \ kill_sandbox, delete_sandbox, start_sandbox, SANDBOX, SANDBOX_CREATE, \ SANDBOX_DELETE, SANDBOX_KILL, SANDBOX_START, SANDBOX_STOP from mysql_gadgets.exceptions import GadgetError PY2 = int(sys.version[0]) == 2 if not PY2: unicode = str # get script name try: _SCRIPT_NAME = os.path.splitext(os.path.split(__file__)[1])[0] except NameError: # Use the default script name if an error occur get __file__ value. _SCRIPT_NAME = 'mysqlprovision' # Force script name if file is renamed to __main__.py for zip packaging. if _SCRIPT_NAME == '__main__': _SCRIPT_NAME = 'mysqlprovision' # Additional supported command names: JOIN = "join-replicaset" LEAVE = "leave-replicaset" START = "start-replicaset" def main(): # Get the provided command command = sys.argv[1] # Read options from stdin data = "" if PY2: stdin = sys.stdin else: # Python3 expects the stdin to be ASCII encoded, while Shell writes the # UTF-8 encoded input. Wrapping the buffer allows to automatially decode # the input. stdin = io.TextIOWrapper(sys.stdin.buffer, encoding='utf-8') while True: line = stdin.readline() if line == ".\n": break data += line shell_options = json.loads(data.decode('utf-8') if PY2 else data) cmd_options = shell_options[0] del shell_options[0] # Setup logging with the appropriate verbosity level. setup_logging(verbosity=int(cmd_options['verbose'])) logging.setLoggerClass(CustomLevelLogger) _LOGGER = logging.getLogger(_SCRIPT_NAME) # Handle keyboard interrupt on password retrieve and command execution. try: # Perform command command_error_msg = "executing operation" try: if command == SANDBOX: sandbox_cmd = cmd_options["sandbox_cmd"] command = '{0} {1}'.format(command, sandbox_cmd) command_error_msg = "executing sandbox operation" if sandbox_cmd == SANDBOX_START: command_error_msg = "starting sandbox" start_sandbox(**cmd_options) elif sandbox_cmd == SANDBOX_CREATE: command_error_msg = "creating sandbox" create_sandbox(**cmd_options) elif sandbox_cmd == SANDBOX_STOP: command_error_msg = "stopping sandbox" stop_sandbox(**cmd_options) elif sandbox_cmd == SANDBOX_KILL: command_error_msg = "killing sandbox" kill_sandbox(**cmd_options) elif sandbox_cmd == SANDBOX_DELETE: command_error_msg = "deleting sandbox" delete_sandbox(**cmd_options) except GadgetError: _, err, _ = sys.exc_info() # the check for mysql.get_classic_session is a hack to prevent # double printing of the DB error if err.cause and u"mysql.get_classic_session" not in unicode(err): _LOGGER.error(u"Error %s: %s: %s", command_error_msg, unicode(err), unicode(err.cause)) else: _LOGGER.error(u"Error %s: %s", command_error_msg, unicode(err)) if _LOGGER.level <= logging.DEBUG: import traceback _LOGGER.debug("%s", traceback.format_exc()) sys.exit(1) except UnicodeEncodeError: _, err, _ = sys.exc_info() _LOGGER.error(u"Unicode error %s: %s. Make sure your locale " u"system settings support UTF-8 to use non-ASCII " u"values.", command_error_msg, unicode(err)) if _LOGGER.level <= logging.DEBUG: import traceback _LOGGER.debug("%s", traceback.format_exc()) sys.exit(1) except Exception: # pylint: disable=broad-except _, err, _ = sys.exc_info() _LOGGER.error(u"Unexpected error %s: %s", command_error_msg, unicode(err)) if _LOGGER.level <= logging.DEBUG: import traceback _LOGGER.debug("%s", traceback.format_exc()) sys.exit(1) # Operation completed with success. sys.exit(0) except KeyboardInterrupt: _LOGGER.error("keyboard interruption (^C) received, stopping...") if os.name == "nt": # Using signal.CTRL_C_EVENT on windows will not set any error code. # Simulate the Unix signal exit code 130 - Script terminated by # Control-C for ngshell to interpreted it as the same way as on # linux. sys.exit(130) else: signal.signal(signal.SIGINT, signal.SIG_DFL) os.kill(os.getpid(), signal.SIGINT) if __name__ == "__main__": # set to True if you want to enable tracing of the Python code # traces are sent to stdout. For debugging only. enable_trace = False if enable_trace: import trace tracer = trace.Trace( ignoredirs=[sys.prefix, sys.exec_prefix], count=0) # run the new command using the given tracer tracer.run('main()') else: main() mysql_gadgets/common/tools.py 0000644 00000043524 15231164450 0012422 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """ This module contains methods for working with mysql server tools. """ # quote() is available in a different module starting from Python 3.3. try: # Import from shlex for Python 3. from shlex import quote except ImportError: # Otherwise, import from pipes module for Python 2. from pipes import quote import shlex import logging import os import signal import socket import subprocess import sys from mysql_gadgets.common.constants import QUOTE_CHAR from mysql_gadgets.common.logger import CustomLevelLogger from mysql_gadgets.exceptions import GadgetError try: import mysqlsh testutil = mysqlsh.globals.testutil except: testutil = None LISTENING_PORT_CHECK_TIMEOUT=5 PY2 = int(sys.version[0]) == 2 logging.setLoggerClass(CustomLevelLogger) _LOGGER = logging.getLogger(__name__) def _add_basedir(search_paths, path_str): """Add a basedir and all known sub directories This method builds a list of possible paths for a basedir for locating special MySQL files like mysqld (mysqld.exe), etc. Note: The resulting paths are append to the list passed by the 'search_paths' parameter. :param search_paths: List of paths to append. The list passed by this parameter is updated by this method. :type search_paths: list :param path_str: The basedir path to append. :param path_str: string """ search_paths.append(path_str) search_paths.append(os.path.join(path_str, "sql")) # for source trees search_paths.append(os.path.join(path_str, "client")) # for source trees search_paths.append(os.path.join(path_str, "share")) search_paths.append(os.path.join(path_str, "scripts")) search_paths.append(os.path.join(path_str, "bin")) search_paths.append(os.path.join(path_str, "libexec")) search_paths.append(os.path.join(path_str, "mysql_utils")) def get_tool_path(basedir, tool, fix_ext=True, required=True, defaults_paths=None, search_path=False, quote=False, check_tool_func=None): """Search for a MySQL tool and return the full path :param basedir: The initial basedir (of a MySQL server) to search. :type basedir: string or None :param tool: The name of the tool to find :type tool: string :param fix_ext: If True (default is True), add .exe if running on Windows. :type fix_ext: boolean :param required: If True (default is True) then an error will be raised if the tool is not found. :type required: boolean :param defaults_paths: Default list of paths to search for the tool. By default (None) an empty list is assumed, i.e. []. :type defaults_paths: list :param search_path: Indicates if the paths specified by the PATH environment variable will be used to search for the tool. By default the PATH will not be searched, i.e. search_path=False. :type search_path: boolean :param quote: if True then the resulting path is surrounded with the OS quotes. :type quote: boolean :param check_tool_func Function to verify the validity of the found tool. This function must take the path of the tool as parameter and return True or False depending if the tool is valid or not (e.g., verify the version). If this check tool function is specified, it will continue searching for the tool in the provided default paths until a valid one is found. By default: None, meaning that it returns the first location found with the tool (without any check). :type check_tool_func function :return: the full path to tool or a list of paths where the tool was found if 'search_all' is set to True, or None if not found and 'required' is set to False. :rtype: string :raises GadgetError: if the tool cannot be found and 'required' is set to True. """ if not defaults_paths: defaults_paths = [] search_paths = [] if quote: if os.name == "posix": quote_char = "'" else: quote_char = '"' else: quote_char = '' if basedir: # Add specified basedir path to search paths _add_basedir(search_paths, basedir) # Search in path from the PATH environment variable if search_path: for path in os.environ['PATH'].split(os.pathsep): search_paths.append(path) if defaults_paths and len(defaults_paths): # Add specified default paths to search paths for path in defaults_paths: search_paths.append(path) else: # Add default MySQL paths to search for tool if os.name == "nt": search_paths.append("C:/Program Files/MySQL/MySQL Server 5.7/bin") search_paths.append("C:/Program Files/MySQL/MySQL Server 8.0/bin") else: search_paths.append("/usr/sbin/") search_paths.append("/usr/local/mysql/bin/") search_paths.append("/usr/bin/") search_paths.append("/usr/local/bin/") search_paths.append("/usr/local/sbin/") search_paths.append("/opt/local/bin/") search_paths.append("/opt/local/sbin/") if os.name == "nt" and fix_ext: tool = "{0}.exe".format(tool) # Search for the tool none_found = True for path in search_paths: norm_path = os.path.normpath(path) if os.path.isdir(norm_path): toolpath = os.path.normpath(os.path.join(norm_path, tool)) if os.path.isfile(toolpath): none_found = False if not check_tool_func or check_tool_func(toolpath): return r"{0}{1}{0}".format(quote_char, toolpath) else: if tool == "mysqld.exe": toolpath = os.path.normpath( os.path.join(norm_path, "mysqld-nt.exe")) if os.path.isfile(toolpath): none_found = False if not check_tool_func or check_tool_func(toolpath): return r"{0}{1}{0}".format(quote_char, toolpath) # Valid tool not found, raise exception or return None. if required: # Distinguish between no tool being found and a valid tool (based on # the check_tool_func parameter) not being found. Use a different # error message and error number. if none_found: raise GadgetError("Cannot find {0}.".format(tool), errno=1) else: raise GadgetError("Cannot find valid {0}.".format(tool), errno=2) return None def get_abs_path(path_string, relative_to=None): """ Get the absolute path for a file This function is used to get the absolute path for the argument provided via the path_string parameter. If the provided path_string is an absolute path, we return it as is, otherwise we will assume it is a relative path relative to the relative_to parameter and use that to calculate the absolute path. :param path_string: Absolute or relative path to which we want to obtain an absolute path. :param path_string: string. :param relative_to: absolute path to a directory or file which will be used as the path to which the path_string argument is relative. :type relative_to: string :return: Absolute path for the path specified in path_string. :rtype: string :raises GadgetError: if path_string is not an absolute path and the provided relative_dir parameter is not an absolute path. """ if path_string[0] == '"' and path_string[-1] == '"': path_string=path_string[1:-1] if os.path.isabs(os.path.expanduser(path_string)): return os.path.expanduser(path_string) else: if not os.path.isabs(relative_to): raise GadgetError("{0} is not a valid absolute path.".format( relative_to)) else: if os.path.isfile(relative_to): relative_to = os.path.dirname(relative_to) return os.path.normpath(os.path.expanduser( os.path.join(relative_to, path_string))) def get_subclass_dict(cls): """Get a dictionary with the subclasses of class 'cls'. This method returns a dictionary with all the classes that inherit from class cls. Note that this method only works for new style classes :param cls: class to which we want to find all subclasses :type cls: class object :return: dictionary whose keys are the names of the subclasses and values are the subclass objects. :rtype: dict """ subclass_dict = {} for subclass in cls.__subclasses__(): subclass_dict[subclass.__name__] = subclass subclass_dict.update(get_subclass_dict(subclass)) return subclass_dict def is_listening(host, port): """Try to check if a given port on a given host is bound and listening. :param host: hostname we want to check :type host: str :param port: port number to check if is bound :type port: int :return: True if port is bound and listening, False otherwise :rtype: bool """ # Socket Initialization if testutil: # testutil will only exist during tests. Outside of tests, this will # throw an exception, which will be ignored and the regular check # is executed. return testutil.is_tcp_port_listening(host, port) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(LISTENING_PORT_CHECK_TIMEOUT) try: s.connect((host, port)) return True except socket.error: return False def is_executable(exec_path): """ Checks if a given path belongs to an executable file :param exec_path: absolute path to the file we want to test is executable. :type exec_path: str :return: True if file exists and is executable, False otherwise :rtype: bool """ _LOGGER.debug("Checking if file '%s' is executable.", exec_path) is_file = os.path.isfile(fs_encode(exec_path)) if not is_file: _LOGGER.debug("File '%s' could not be found.", exec_path) return False elif os.access(fs_encode(exec_path), os.X_OK): _LOGGER.debug("File '%s' exists and is executable.", exec_path) return True else: _LOGGER.debug("File '%s' exists but is not executable.", exec_path) return False def run_subprocess(cmd_str, **kwargs): """Runs the provided command in a subprocess. This method takes the provided argument cmd_str, splits it into a argument list and then feeds it to a subprocess.Popen constructor along with any extra keyword argument provided. It return the subprocess.Popen instance. :param cmd_str: string with the command to be executed. :type cmd_str: str :param kwargs: dictionary of extra parameters to be passed to the subprocess.Popen call :type kwargs: dict :return: the subprocess.Popen instance :rtype: subprocess.Popen """ _LOGGER.debug("Spawning subprocess with command '%s'", cmd_str) if os.name == "nt": if PY2: # on windows we can use the string directly cmd_list = cmd_str.encode('mbcs') else: cmd_list = cmd_str # create process without a console window (also detaching from our console) CREATE_NO_WINDOW = 0x08000000 # create process in a new group, otherwise pressing CTRL-C in the # current console will kill the created process kwargs['creationflags'] = kwargs.get( 'creationflags', 0) | subprocess.CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW kwargs['close_fds'] = kwargs.get('close_fds', False) # on Windows 'close_fds' closes all descriptors, including standard # input/output/error, redirection in this case cannot be used if kwargs['close_fds'] and sys.version_info[:2] < (3, 7): kwargs['stdin'] = None kwargs['stdout'] = None kwargs['stderr'] = None else: if PY2: # shlex.split() does not fully support UTF-8. enc_cmd_str = cmd_str.encode('utf-8') else: enc_cmd_str = cmd_str cmd_list = map(lambda s: s, shlex.split(enc_cmd_str)) proc = subprocess.Popen(cmd_list, **kwargs) return proc def stop_process_with_pid(pid, force=False): """Terminates or kills a process with a given pid. This method attempts to stop a process with the provided pid. If force parameter is False, this it attempts to gracefully terminate the process. Otherwise, if force parameter is True it attempts to forcefully terminate the process. :param pid: pid of the process we want to terminate/kill. :type pid: int :param force: If True process is forcefully killed, otherwise it is gracefully terminated. :type force: bool :raises GadgetError: if unable to kill or terminate the process. """ error_msg = "Unable to {0} process '{1}': '{2}'" if force: if os.name == "nt": # windows doesn't have sigkill, we need to use taskkill # to terminate a process. kill_proc = run_subprocess("taskkill /PID {0} /F".format(pid), shell=False, stderr=subprocess.PIPE, universal_newlines=True) _, err = kill_proc.communicate() if kill_proc.returncode: raise GadgetError(error_msg.format("kill", pid, str(err))) else: # posix try: # send SIGKILL # pylint: disable=E1101 os.kill(pid, signal.SIGKILL) except OSError as err: raise GadgetError(error_msg.format("kill", pid, str(err))) else: try: # send SIGTERM os.kill(pid, signal.SIGTERM) except OSError as err: raise GadgetError(error_msg.format("terminate", pid, str(err))) def shell_quote(param): """Quote the string parameter from the shell command line. Quote the specified string parameter, returning a shell-escaped version of the string that can be safely used in a shell command line. Note: Quoting is compatible with UNIX shells and with shlex.split(), for Windows only the parameter is only surrounded with double quotes ("). :param param: Parameter to quote. :type param: string :return: A shell-escaped version of the specified string parameter. """ if os.name == "posix": return quote(param) else: return u"{0}{1}{0}".format(QUOTE_CHAR, param) def fs_encode(value): """ Encode the value according to the file system default codec. This function is used to encode non-ascii strings (e.g., paths) correctly, according to the default system codec. By default, on non-windows platforms 'utf-8' is used (supported), and on Windows 'mbcs' (ANSI) is used. Not using the right codec will result in incorrect characters, and failure of file system operations for paths with non-ascii characters. NOTE: On non-windows platforms, even if locale are not set (LC_CTYPE), 'utf-8' can be used since it is expected to be supported, otherwise an UnicodeEncodeError will be raised. For this reason, sys.getfilesystemencoding() is not used to automatically get the encoding, "forcing" the use of the expected supported codec independently of the current locale settings. :param value: value (non-ascii) to encode. :return: encoded value according to the default system codec. """ if PY2: if os.name == "nt": return value.encode('mbcs') else: return value.encode('utf-8') else: return value def fs_decode(value): """ Decode the value according to the file system default codec. This function is used to decode non-ascii strings correctly, according to the default system codec. By default, on non-windows platforms 'utf-8' is used (supported), and on Windows 'mbcs' (ANSI) is used. Not using the right codec will result in incorrect characters. :param value: value (non-ascii) to decode. :return: decoded value according to the default system codec. """ if PY2: if os.name == "nt": return value.decode('mbcs') else: return value.decode('utf-8') else: return value mysql_gadgets/common/server.py 0000644 00000216443 15231164450 0012572 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """ This module contains an abstraction of a MySQL server object used by multiple gadgets. It also contains methods and functions for common server operations. """ import logging import os import random import re import socket import threading import time import sys import subprocess import mysqlsh from mysql_gadgets import MIN_MYSQL_VERSION, MAX_MYSQL_VERSION from mysql_gadgets.exceptions import (GadgetCnxInfoError, GadgetCnxError, GadgetQueryError, GadgetServerError, GadgetError) from mysql_gadgets.common.connection_parser import (parse_connection, hostname_is_ip, clean_IPv6,) from mysql_gadgets.common.logger import CustomLevelLogger from mysql_gadgets.common.tools import (get_abs_path, is_executable, run_subprocess, shell_quote) CR_SERVER_LOST = 2013 ER_OPTION_PREVENTS_STATEMENT = 1290 _FOREIGN_KEY_SET = "SET foreign_key_checks = {0}" _AUTOCOMMIT_SET = "SET AUTOCOMMIT = {0}" _GTID_ERROR = ("The server {host}:{port} does not comply to the latest GTID " "feature support. Errors:") logging.setLoggerClass(CustomLevelLogger) _LOGGER = logging.getLogger(__name__) class Secret(object): def __init__(self, s): super(Secret, self).__init__() self.value = s def __str__(self): return str(self.value) def __repr__(self): return repr(self.value) class Query(object): def __init__(self, query_string, *params): super(Query, self).__init__() self._query_string = query_string self._params = params def extend(self, query_string, *params): self._query_string += " " + query_string self._params.extend(params) def query(self): return self._query_string def params(self, mask=False): if mask: return tuple(map(lambda x: "<secret>" if isinstance(x, Secret) else x, self._params)) return tuple(map(lambda x: x.value if isinstance(x, Secret) else x, self._params)) def __str__(self): if self._params: return "{0}, with params: {1}".format(self.query(), self.params(True)) return self.query() def log(self): return self.__str__() def _to_str(value, charset="utf-8"): """Cast value to str except when None :param value: Value to be cast to str :type value: bytearray :param charset: the charset to decode the value. Used only on python 3 :type charset: string :return: value as string instance or None. :rtype: string or None """ if sys.version_info > (3, 0): # for Python 3 use bytes instead of str return None if value is None else bytes(value).decode(charset) else: return None if value is None else str(value) def get_mysqld_version(mysqld_path): """Get the version of the mysql server through the mysqld executable. :param mysqld_path: absolute path to the mysqld executable :type mysqld_path: str :return: tuple with with major, minor and release number and version string :rtype tuple((int, int, int), str) """ cmd = u"{0} --version".format(shell_quote(mysqld_path)) version_proc = run_subprocess(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, universal_newlines=True) output, error = version_proc.communicate() match = re.match(r'^.*mysqld.*?\s+(Ver\s+(\d+\.\d+(?:\.\d+)*).*)', output) if match: return tuple(int(n) for n in match.group(2).split('.')), match.group(1) else: error_msg = '' if error: error_msg = " Error executing '{0}': {1}".format(cmd, error) raise GadgetError( "Unable to parse version output '{0}' from mysqld executable '{1}'" ".{2}".format(output, mysqld_path, error_msg)) def is_valid_mysqld(mysqld_path): """Check if the provided mysqld is valid. Check the version of the given mysqld and return True if valid and False otherwise. :param mysqld_path: Path to mysqld to check. :type mysqld_path: string :return: True if the provided mysqld is valid, or False otherwise. :rtype: bool """ # Check if mysqld is executable and version is valid. if is_executable(mysqld_path): mysqld_ver, _ = get_mysqld_version(mysqld_path) if MIN_MYSQL_VERSION <= mysqld_ver < MAX_MYSQL_VERSION: _LOGGER.debug("Valid mysqld found with version %s: '%s'", ".".join(str(num) for num in mysqld_ver), mysqld_path) return True # mysqld is not valid. _LOGGER.debug("Invalid mysqld version (or not executable): '%s'", mysqld_path) return False class MySQLUtilsCursorResult(object): def __init__(self, result): self._result = result self.with_rows = result.has_data() self.column_names = result.column_names def fetchone(self): row = self._result.fetch_one() if row: return tuple(str(row[i]) if row[i] is not None else 'NULL' for i in range(row.length)) return None def fetchall(self): rows = [] row = self.fetchone() while row: rows.append(row) row = self.fetchone() return rows def close(self): pass def get_connection_dictionary(conn_info, ssl_dict=None): """Get the connection dictionary. Convert the given connection information into a dictionary. The method accepts one of the following types for the connection information: - dictionary containing connection information including: (user, passwd, host, port, socket) - connection string in the form: user:pass@host:port:socket - an instance of the Server class :param conn_info: Connection information to be converted. :type conn_info: dictionary or Server object or string :param ssl_dict: A dictionary with the ssl options. :type ssl_dict: dictionary :return dictionary with connection information (user, passwd, host, port, socket). :rtype: dictionary """ if conn_info is None: return conn_info if isinstance(conn_info, dict): # Not update conn_info if already has any ssl certificate. if (ssl_dict is not None and not (conn_info.get("ssl_ca", None) or conn_info.get("ssl_cert", None) or conn_info.get("ssl_key", None) or conn_info.get("ssl", None))): conn_info.update(ssl_dict) conn_val = conn_info elif isinstance(conn_info, Server): # get server's dictionary conn_val = conn_info.get_connection_values() elif isinstance(conn_info, str): # parse the string conn_val = parse_connection(conn_info, options=ssl_dict) else: raise GadgetCnxInfoError("Cannot determine connection information" " type.") return conn_val def check_hostname_alias(server1_cnx_values, server2_cnx_values): """Check to see if the servers are the same machine by host name. :param server1_cnx_values: connection values for server 1 :type server1_cnx_values: dictionary :param server2_cnx_values: connection values for server 2 :type server2_cnx_values: dictionary :return: True if both servers are the same otherwise False. :rtype: boolean """ server1 = Server({'conn_info': server1_cnx_values}) server2 = Server({'conn_info': server2_cnx_values}) return (server1.is_alias(server2.host) and int(server1.port) == int(server2.port)) def get_server(server_info, ssl_dict=None, connect=True): """Get a server instance from server connection information | string or a Server instance. The method accepts one of the following types for server_info: - dictionary containing connection information including: (user, passwd, host, port, socket) - connection string in the form: user:pass@host:port:socket or login-path:port:socket - an instance of the Server class :param server_info: Connection information :type server_info: dict | Server | str :param ssl_dict: A dictionary with the ssl certificates :type ssl_dict: dict :param connect: Attempt to connect to the server :type connect: boolean :raise GadgetCnxInfoError: if the connection information on server_info could not be parsed :raise GadgetServerError: if a connection fails. @returns a Server instance :rtype: Server """ if server_info is None: return server_info if isinstance(server_info, dict) and 'host' in server_info: # Don't update server_info if already has any ssl certificate. if (ssl_dict is not None and not (server_info.get("ssl_ca", None) or server_info.get("ssl_cert", None) or server_info.get("ssl_key", None) or server_info.get("ssl", None))): server_info.update(ssl_dict) options = {"conn_info": server_info} server = Server(options) elif isinstance(server_info, Server): server = server_info elif isinstance(server_info, str): # parse the string conn_val = parse_connection(server_info, options=ssl_dict) options = {"conn_info": conn_val} server = Server(options) else: raise GadgetCnxInfoError("Cannot determine connection information" " type.") if connect: server.connect() return server def generate_server_id(strategy='RANDOM'): """Generate a server ID based on the provided strategy. Two strategies are supported to generate a server_id: TIME and RANDOM. The TIME strategy allow to generate a server_id based on the current timestamp, ensuring that a unique server_id can be generated every second during approximately 31 years (until 2038, epoch end for the time.time() function in Unix). If different server_ids are generated during the same second they will all have the same value. Minimum value generated is 1 and maximum 999999999. The RANDOM strategy allow to generate a pseudo-random server_id, with a value between 1 and 4294967295. Two generated values have a low probability of being the same (independently of the time they are generated). Minimum value generated is 1 and maximum 4294967295. For the random generation (assuming random generation is uniformly) the probability of the same value being generated is given by: P(n, t) = 1 - t!/(t^n * (t-n)!) where t is the total number of different values that can be generated, and n is the number of values that are generated. In this case, t = 4294967295 (max number of values that can be generated), and for example the probability of generating the same id for 15, 100, and 1000 servers (n=15, n=100, and n=1000) is approximately: P(15, 4294967295) = 2.44 * 10^-8 (0.00000244 %) P(100, 4294967295) = 1.15 * 10^-6 (0.000115 %) P(1000, 4294967295) = 1.16 * 10^-4 (0.0116 %) Note: Zero is not a valid sever_id. :param strategy: Strategy used to generate the server id value. Supported values: 'RANDOM' to generate a random ID and 'TIME' to generate the ID based on the current timestamp. By default: 'RANDOM'. :type strategy: String :return: The generated server_id. :rtype: string :raises GadgetError: If an unsupported strategy is specified. """ strategy = strategy.upper() if strategy == 'RANDOM': # Generate random int between the min valid and max allowed value. # Note: 0 is not a valid server_id. return str(random.randint(1, 4294967295)) elif strategy == 'TIME': # Only the nine last digits from the time can be used to avoid # generating a value higher than the max allowed server_id value. server_id = str(int(time.time()))[-9:] # 0 is not a valid server_id, in that case return 1 (1 second later). if int(server_id) == 0: return '1' return server_id else: raise GadgetError("Invalid strategy used to generate the server_id " "(supported values: 'RANDOM' or 'TIME'): " "{0}.".format(strategy)) # pylint: disable=too-many-public-methods class Server(object): """The Server class can be used to connect to a running MySQL server. It provides several features, such as: - connect/disconnect to the server - get the server version - Retrieve a server variable - Execute a query, commit, and rollback - Return list of all databases - Read SQL statements from a file and execute - check for specific plugin support - etc. """ def __init__(self, options): """Constructor The constructor accepts one of the following types for the connection information provided trough the options parameter, more specifically for the options['conn_info']: - dictionary containing connection information including: (user, passwd, host, port, socket) - connection string in the form: user:pass@host:port:socket - an instance of the Server class :param options: Options and definitions used to create the Server object. Supported key values: conn_info a dictionary containing connection information (user, passwd, host, port, socket) role Name or role of server (e.g., server, master) verbose print extra data during operations (optional) default value = False charset Default character set for the connection. (default None) :type options: dictionary """ if options is None: options = {} if options.get("conn_info") is None: raise GadgetCnxInfoError( "Server connection information missing. The option parameter" "must contain a 'conn_info' entry.") self.verbose = options.get("verbose", False) self.db_conn = None self.host = None self.role = options.get("role", "Server") self.has_ssl = False conn_values = get_connection_dictionary(options.get("conn_info")) try: self.host = conn_values["host"] self.user = conn_values["user"] self.passwd = conn_values["passwd"] \ if "passwd" in conn_values else None self.socket = conn_values["unix_socket"] \ if "unix_socket" in conn_values else None self.port = 3306 if conn_values["port"] is not None: self.port = int(conn_values["port"]) self.charset = options.get("charset", conn_values.get("charset", None)) # Optional values self.ssl_ca = conn_values.get('ssl_ca', None) self.ssl_cert = conn_values.get('ssl_cert', None) self.ssl_key = conn_values.get('ssl_key', None) self.ssl = conn_values.get('ssl', False) if self.ssl_cert or self.ssl_ca or self.ssl_key or self.ssl: self.has_ssl = True except KeyError as err: raise GadgetCnxInfoError( "Server connection dictionary format not recognized. " "Mandatory value missing ({0}): " "{1}".format(str(err), conn_values)) self.connect_error = None # Set to TRUE when foreign key checks are ON. Check with # foreign_key_checks_enabled. self.fkeys = None self.autocommit = True # Set autocommit to True by default. self.read_only = False self.aliases = set() self.grants_enabled = None self._version = None self._version_full = None @classmethod def from_server(cls, server, conn_info=None): """ Create a new server instance from an existing one. Factory method that will allow the creation of a new server instance from an existing server. :param server: Source object used to create the new server. It must be an instance of the Server class or a subclass. :type server: Server object :param conn_info: Connection information for the new server. If provided this information will overwrite the one from the source server. :type conn_info: dictionary :return: An new instance of the Server class based on the provided source server snf (optional) connection information.. :rtype: Server object """ if isinstance(server, Server): options = {"role": server.role, "verbose": server.verbose, "charset": server.charset} if conn_info is not None and isinstance(conn_info, dict): options["conn_info"] = conn_info else: options["conn_info"] = server.get_connection_values() return cls(options) else: raise TypeError("The server argument's type is neither Server nor " "a subclass of Server") def is_alive(self): """Determine if connection to server is alive. :returns: True if server is alive (and responding) or False if an error occurred when trying to connect to the server. :rtype: boolean """ res = True try: if self.db_conn is None: res = False else: # ping and is_connected only work partially, try exec_query # to make sure connection is really alive retval = self.db_conn.is_open() if retval: self.exec_query("SELECT 1") else: res = False except Exception: # pylint: disable=W0703 res = False return res def _update_alias(self, ip_or_hostname, suffix_list): """Update list of aliases for the given IP or hostname. Gets the list of aliases for host *ip_or_hostname*. If any of them matches one of the server's aliases, then update the list of aliases (self.aliases). It also receives a list (tuple) of suffixes that can be ignored when checking if two hostnames are the same. :param ip_or_hostname: IP or hostname to test. :type ip_or_hostname: string :param suffix_list: Tuple with list of suffixes that can be ignored. :type suffix_list: list :returns: True if ip_or_hostname is a server alias, otherwise False. :rtype: boolean """ host_or_ip_aliases = self._get_aliases(ip_or_hostname) host_or_ip_aliases.add(ip_or_hostname) # Check if any of aliases matches with one the servers's aliases common_alias = self.aliases.intersection(host_or_ip_aliases) if common_alias: # There are common aliases, host is the same self.aliases.update(host_or_ip_aliases) return True else: # Check with and without suffixes no_suffix_server_aliases = set() no_suffix_host_aliases = set() for suffix in suffix_list: # Add alias with and without suffix from self.aliases for alias in self.aliases: if alias.endswith(suffix): host, _ = alias.rsplit('.', 1) no_suffix_server_aliases.add(host) no_suffix_server_aliases.add(alias) # Add alias with and without suffix from host_aliases for alias in host_or_ip_aliases: if alias.endswith(suffix): host, _ = alias.rsplit('.', 1) no_suffix_host_aliases.add(host) no_suffix_host_aliases.add(alias) # Check if there is any alias in common common_alias = no_suffix_host_aliases.intersection( no_suffix_server_aliases) if common_alias: # Same host, so update self.aliases self.aliases.update( no_suffix_host_aliases.union(no_suffix_server_aliases) ) return True return False def _get_aliases(self, host): """Gets the aliases for the given host. :param host: the host name or IP :type host: string :return: aliases for the given host. :rtype: list """ aliases = set([clean_IPv6(host)]) if hostname_is_ip(clean_IPv6(host)): # IP address try: my_host = socket.gethostbyaddr(clean_IPv6(host)) aliases.add(my_host[0]) # socket.gethostbyname_ex() does not work with ipv6 if (not my_host[0].count(":") < 1 or not my_host[0] == "ip6-localhost"): host_ip = socket.gethostbyname_ex(my_host[0]) else: addrinfo = socket.getaddrinfo(my_host[0], None) host_ip = ([socket.gethostbyaddr(addrinfo[0][4][0])], [fiveple[4][0] for fiveple in addrinfo], [addrinfo[0][4][0]]) except (socket.gaierror, socket.herror, socket.error) as err: host_ip = ([], [], []) if self.verbose: _LOGGER.warning( "IP lookup by address failed for %s, reason: %s", host, err.strerror) else: try: # server may not really exist. host_ip = socket.gethostbyname_ex(host) except (socket.gaierror, socket.herror, socket.error) as err: if self.verbose: _LOGGER.warning( "hostname: %s may not be reachable,reason: %s", host, err.strerror) return aliases aliases.add(host_ip[0]) addrinfo = socket.getaddrinfo(host, None) local_ip = None error = None for addr in addrinfo: try: local_ip = socket.gethostbyaddr(addr[4][0]) break except (socket.gaierror, socket.herror, socket.error) as err: error = err if local_ip: host_ip = ([local_ip[0]], [fiveple[4][0] for fiveple in addrinfo], [addrinfo[0][4][0]]) else: host_ip = ([], [], []) if self.verbose: _LOGGER.warning( "IP lookup by name failed for %s, reason: %s", host, error.strerror) aliases.update(set(host_ip[1])) aliases.update(set(host_ip[2])) return aliases def is_alias(self, host_or_ip): """Determine if the given host is an alias of the server host. :param host_or_ip: host or IP address to check. :type host_or_ip: string :returns: True is the given host or IP address is an alias of the server host, otherwise False. :rtype: boolean """ # List of possible suffixes suffixes = ('.local', '.lan', '.localdomain') host_or_ip = clean_IPv6(host_or_ip.lower()) # for quickness, verify in the existing aliases, if they exist. if self.aliases: if host_or_ip.lower() in self.aliases: return True else: # get the alias for the given host_or_ip return self._update_alias(host_or_ip, suffixes) # no previous aliases information # First, get the local information hostname_ = socket.gethostname() try: local_info = socket.gethostbyname_ex(hostname_) local_aliases = set([local_info[0].lower()]) # if dotted host name, take first part and use as an alias try: local_aliases.add(local_info[0].split('.')[0]) except Exception: # pylint: disable=W0703 pass local_aliases.update(['127.0.0.1', 'localhost', '::1', '[::1]']) local_aliases.update(local_info[1]) local_aliases.update(local_info[2]) local_aliases.update(self._get_aliases(hostname_)) except (socket.herror, socket.gaierror, socket.error) as err: if self.verbose: _LOGGER.warning("Unable to find aliases for hostname '%s', " "reason: %s", hostname_, str(err)) # Try with the basic local aliases. local_aliases = set(['127.0.0.1', 'localhost', '::1', '[::1]']) # Get the aliases for this server host self.aliases = self._get_aliases(self.host) # Check if this server is local for host in self.aliases.copy(): if host in local_aliases: # Is local then save the local aliases for future. self.aliases.update(local_aliases) break # Handle special suffixes in hostnames. for suffix in suffixes: if host.endswith(suffix): # Remove special suffix and attempt to match with local # aliases. host, _ = host.rsplit('.', 1) if host in local_aliases: # Is local then save the local aliases for future. self.aliases.update(local_aliases) break # Check if the given host_or_ip is alias of the server host. if host_or_ip in self.aliases: return True # Check if any of the aliases of ip_or_host is also an alias of the # host server. return self._update_alias(host_or_ip, suffixes) def user_host_exists(self, user_name, host_name): """Check if the 'user_name'@'host_name' account exists. This function checks if the specified 'user_name' and 'host_name' match an existing account on the server, considering the respective parts of the account name 'user_name'@'host_name'. Wildcard (%) matches are also taken into consideration for the 'host_name' part of the account name. For example, if an account 'myname'@'%' exists then any user with the name 'myname' will match the account independently of the host name. :param user_name: user name of the account. :type user_name: string :param host_name: hostname or IP address of the account. Note: wildcard '%' can be used. :type host_name: string :return: True if the given 'user_name' and 'host_name' match an existing account, otherwise False. :rtype: boolean :raise GadgetServerError: If an error occurs getting the user accounts information. """ res = self.exec_query(Query("SELECT host FROM mysql.user WHERE user = ? " "AND ? LIKE host", user_name, host_name)) if res: return True return False def get_connection_values(self): """Return a dictionary of connection values for the server. :return: Return the connection information for the server. :rtype: dictionary """ conn_vals = { "user": self.user, "host": self.host } if self.passwd: conn_vals["passwd"] = self.passwd if self.socket: conn_vals["unix_socket"] = self.socket if self.port: conn_vals["port"] = self.port if self.ssl_ca: conn_vals["ssl_ca"] = self.ssl_ca if self.ssl_cert: conn_vals["ssl_cert"] = self.ssl_cert if self.ssl_key: conn_vals["ssl_key"] = self.ssl_key if self.ssl: conn_vals["ssl"] = self.ssl return conn_vals def connect(self): """Connect to the server. Attempts to connect to the server according to its connection parameters. Note: This method must be called before executing statements. :raise GadgetServerError: if an error occurs during the connection. """ try: self.db_conn = self.get_connection() if self.ssl: res = self.exec_query("SHOW STATUS LIKE 'Ssl_cipher'") if res[0][1] == '': raise GadgetCnxError("Can not encrypt server connection.") except GadgetServerError: # Reset any previous value if the connection cannot be established, # before raising an exception. This prevents the use of a broken # database connection. self.db_conn = None raise self.connect_error = None self.read_only = self.show_server_variable("READ_ONLY")[0][1] def get_connection(self): """Return a new connection to the server. Attempts to connect to the server according to its connection parameters and returns a connection object. :return: The resulting MySQL connection object. :rtype: MySQLConnection object :raise GadgetCnxError: if an error occurred during the server connection process. """ try: parameters = { 'user': self.user, 'host': self.host, 'port': self.port, } if self.socket and os.name == "posix": parameters['socket'] = self.socket if self.passwd and self.passwd != "": parameters['password'] = self.passwd parameters['host'] = parameters['host'].replace("[", "") parameters['host'] = parameters['host'].replace("]", "") # Add SSL parameters ONLY if they are not None if self.ssl_ca is not None: parameters['ssl-ca'] = self.ssl_ca if self.ssl_cert is not None: parameters['ssl-cert'] = self.ssl_cert if self.ssl_key is not None: parameters['ssl-key'] = self.ssl_key if self.ssl == False: parameters['ssl-mode'] = "DISABLED" # The ca certificate is verified only if the ssl option is also # specified. if self.ssl and parameters['ssl-ca']: parameters['ssl-mode'] = "VERIFY_CA" if not "ssl-mode" in parameters: parameters['ssl-mode'] = "PREFERRED" db_conn = mysqlsh.mysql.get_classic_session(parameters) # Return MySQL connection object. return db_conn except mysqlsh.DBError as err: _LOGGER.debug("Connector Error: %s", err) raise GadgetCnxError( err.args[1], err.args[0], cause=err, server=self) except AttributeError as err: # Might be raised by mysql.connector.connect() raise GadgetCnxError(str(err), cause=err, server=self) def disconnect(self): """Disconnect from the server. """ if self.db_conn is None: raise GadgetCnxError("Cannot disconnect from a not connected" "server. You must use connect() first.") try: self.db_conn.close() except mysqlsh.DBError: # No error expected even if already disconnected, anyway ignore it. pass def get_version(self, full=False): """Return version number of the server. Get the server version. The respective instance variable is set with the result after querying the server the first time. The version is immediately returned when already known, avoiding querying the server at each time. :param full: If True return the full version information including suffixes (e.g., 5.7.14-log), otherwise only the major, minor and release number (e.g., 5.7.14) are returned. By default: False. :type full: boolean :return: List of version numbers(ints) if full = False or a string if full = True or None if an error occurs when trying to get the version information from the server. :rtype: string or int list or None """ # Return the local version value if already known. if self._version: if full: return self._version_full else: return self._version # Query the server for its version. try: res = self.show_server_variable("VERSION") if res: self._version_full = res[0][1] match = re.match(r'^(\d+\.\d+(\.\d+)*).*$', self._version_full.strip()) if match: self._version = [int(x) for x in match.group(1).split('.')] # Ensure a 3 elements list self._version = (self._version + [0])[:3] except (GadgetCnxError, GadgetQueryError): # Ignore errors and return version initialized with None. pass if full: return self._version_full else: return self._version def check_version_compat(self, t_major, t_minor, t_rel): """Checks version of the server against requested version. This method can be used to check for version compatibility. :param t_major: target server version (major) :type t_major: str or int :param t_minor: target server version (minor) :type t_minor: str or int :param t_rel: target server version (release) :type t_rel: str or int :return: True if server version is greater or equal (>=) than the specified version. False if server version is lower (<) than the specified version. :rtype: boolean """ version = self.get_version() if version is not None: return version >= [int(t_major), int(t_minor), int(t_rel)] else: return False def exec_query(self, query, options=None, exec_timeout=0): """Execute a query and return result. This is the singular method to execute queries. It should be the only method used as it contains critical error code to catch the issue with mysql.connector throwing an error on an empty result set. Notes: - It handles exception if query fails. - If 'fetch' is False in the options, the method returns the cursor instance. - By default a commit is performed at the end, unless 'commit' is set to False in the options. :param query: Query object or string with the SQL statement to execute. :type query: Query :type query: str :param options: Options to control the execution of the statement. The follow values are supported: columns Add column headings as first row. By default, False. fetch Execute the fetch as part of the operation and use a buffered cursor. By default, True. raw If True use a buffered raw cursor, meaning that all returned values are strings (i.e., not converted to the corresponding Python type). By default, True. commit Perform a commit (if needed) automatically at the end. By default, True. :type options: dictionary :param exec_timeout: Timeout value in seconds to kill the query execution if exceeded. Value must be greater than zero for this feature to be enabled. By default 0, meaning that the query will not be killed. :type exec_timeout: integer :return: List of rows (tuples) or Cursor with the result of the query. :rtype: list of tuples or Cursor object :raise GadgetCnxError: If an error occurs with the server connection or creating the cursor. :raise GadgetQueryError: If an error occurs when excuting the statement (query), fetching results or committing changes. """ if options is None: options = {} columns = options.get('columns', False) fetch = options.get('fetch', True) raw = options.get('raw', True) do_commit = options.get('commit', True) if isinstance(query, str): query = Query(query) # Guard for connect() prerequisite if not self.db_conn: raise GadgetCnxError( "You must call connect before executing a query.", server=self) # Execute query, handling parameters. q_killer = None try: if exec_timeout > 0: if query.params(): raise NotImplementedError() # Spawn thread to kill query if timeout is reached. # Note: set it as daemon to avoid waiting for it on exit. q_killer = QueryKillerThread(self, query.query(), exec_timeout) q_killer.daemon = True q_killer.start() # Execute query. cur = None _LOGGER.debug("MySQL query: " + query.log()) if query.params(): cur = MySQLUtilsCursorResult( self.db_conn.run_sql(query.query(), query.params())) else: cur = MySQLUtilsCursorResult( self.db_conn.run_sql(query.query())) except mysqlsh.DBError as err: if cur: cur.close() if err.args[0] == CR_SERVER_LOST and exec_timeout > 0: # If the connection is killed (because the execution timeout is # reached), then it attempts to re-establish it (to execute # further queries) and raise a specific exception to track this # event. # CR_SERVER_LOST = Errno 2013 Lost connection to MySQL server # during query. self.connect() raise GadgetQueryError("Timeout executing query", query.log(), errno=err.args[0], cause=err, server=self) else: raise GadgetQueryError("Query failed. {0}".format(err), query.log(), errno=err.args[0], cause=err, server=self) except Exception as err: if cur: cur.close() raise GadgetQueryError("Unknown error: {0}".format(err), query.log(), errno=0, cause=err, server=self) finally: # Stop query killer thread if alive. if q_killer and q_killer.is_alive(): q_killer.stop() # Fetch rows (only if available or fetch = True). if cur.with_rows: if fetch or columns: try: results = cur.fetchall() if columns: col_headings = cur.column_names col_names = [] for col in col_headings: col_names.append(col) results = col_names, results except mysqlsh.DBError as err: raise GadgetQueryError( "Error fetching all query data: {0}".format(err), query.log(), errno=err.args[0], cause=err, server=self) finally: cur.close() return results else: # Return cursor to fetch rows elsewhere (fetch = false). return cur else: # No results (not a SELECT) try: if do_commit: self.db_conn.run_sql("commit") except mysqlsh.DBError as err: raise GadgetQueryError( "Error performing commit: {0}".format(err), query.log(), errno=err.args[0], cause=err, server=self) finally: cur.close() return cur def commit(self): """Perform a COMMIT. :raise GadgetCnxError: If connection was not previously established. """ # Guard for connect() prerequisite if not self.db_conn: raise GadgetCnxError( "You must call connect before commit.", server=self) self.db_conn.run_sql("commit") def rollback(self): """Perform a ROLLBACK. :raise GadgetCnxError: If connection was not previously established. """ # Guard for connect() prerequisite if not self.db_conn: raise GadgetCnxError( "You must call connect before rollback.", server=self) self.db_conn.run_sql("rollback") def show_server_variable(self, variable): """Get the variable information using SHOW VARIABLES statement. Return one or more rows from the SHOW VARIABLES statement according to the specified variable parameter. :param variable: The variable name or wildcard string to match the returned variable information. :type variable: string :returns: Variables names and respective value matching the specified variable parameter. :rtype: list of tuples :raise GadgetServerError: If an error occur getting the variable value. """ return self.exec_query(Query("SHOW VARIABLES LIKE ?", variable)) def set_variable(self, var_name, var_value, var_type="session"): """Set server variable using the SET statement. This function sets the value of system variables using the SET statement. :param var_name: Name of the variable to set. :type var_name: str :param var_value: value to which we want to set the variable. :type var_value: str :param var_type: Type of the variable ('session', 'global', 'persist' 'persist_only'). By default session is used. :type var_type: str :raises GadgetDBError: if an invalid var_type is provided as argument. """ if var_type.lower() in ('global', 'session', 'persist', 'persist_only'): var_type = '{0}.'.format(var_type) # Add dot (.) else: raise GadgetError("Invalid variable type: {0}. Supported types: " "'global' and 'session'.".format(var_type)) # Execute SET @@var_type.var_name. self.exec_query("SET @@{0}{1}={2}".format(var_type, var_name, var_value)) def select_variable(self, var_name, var_type=None): """Get server system variable value using SELECT statement. This function displays the value of system variables using the SELECT statement. This can be used as a workaround for variables with very long values, as SHOW VARIABLES is subject to a version-dependent display-width limit. Note: Some variables may not be available using SELECT @@var_name, in such cases use SHOW VARIABLES LIKE 'var_name'. :param var_name: Name of the variable to display. :type var_name: string :param var_type: Type of the variable ('session' or 'global'). By default None (no type is used), meaning that the session value is returned if it exists and the global value otherwise. :type var_type: 'session', 'global', '', or None :return: value for the given server system variable. :rtype: string :raise GadgetServerError: If an unsupported variable type is specified. :raise GadgetQueryError: If the variable does not exist in the server. """ if var_type is None: var_type = '' elif var_type.lower() in ('global', 'session', ''): var_type = '{0}.'.format(var_type) # Add dot (.) else: raise GadgetServerError( "Invalid variable type: {0}. Supported types: " "'global' and 'session'.".format(var_type), server=self) # Execute SELECT @@[var_type.]var_name. # Note: An error is issued if the given variable is not known. res = self.exec_query("SELECT @@{0}{1}".format(var_type, var_name)) return res[0][0] def has_default_value(self, var_name): """Check if the variable has the default value or was already change. Return a boolean value indicating if the variable is set with the compiled default value (true), or if it was already changed explicitly by the user somehow (false), i.e. variable changed with the SET statement, a command line option, or the configuration file. NOTE: This method requires the performance_schema to be enabled. :param var_name: Name of the target variable to check. :type var_name: string :return: True if the variable value is the compiled default one, or False otherwise (meaning that the variable value was already changed by the user). :rtype: boolean """ res = self.exec_query(Query("SELECT variable_source " "FROM performance_schema.variables_info " "WHERE variable_name=?", var_name)) if res[0][0] == 'COMPILED': return True else: return False def flush_logs(self, log_type=None): """Execute the FLUSH [log_type] LOGS statement. Reload internal logs cache and closes and reopens all log files, or only of the specified log_type. Note: The log_type option is available starting from MySQL 5.5.3. :param log_type: Type of the log files to be flushed. Supported values: BINARY, ENGINE, ERROR, GENERAL, RELAY, SLOW. :type log_type: string :raise GadgetServerError: If an error occurs when executing the FLUSH LOGS statement. """ if log_type: self.exec_query("FLUSH {0} LOGS".format(log_type)) else: self.exec_query("FLUSH LOGS") def supports_gtid(self): """Check if the server supports GTIDs. :return: True if GTID is supported and turned on and False if supported but not enabled. :rtype: boolean :raise GadgetServerError: If GTID is not supported or the GTID mode cannot be obtained. """ # Check server for GTID support version_ok = self.check_version_compat(5, 6, 5) if not version_ok: raise GadgetServerError("GTIDs are only supported starting from" "MySQL 5.6.5", server=self) try: res = self.exec_query("SELECT @@GLOBAL.GTID_MODE") except (GadgetCnxError, GadgetQueryError) as err: raise GadgetServerError( "Unable to get GTID_MODE value: {0}".format(err.errmsg), cause=err, server=self) # Return result if res[0][0] == 'ON': return True elif res[0][0] == 'OFF': return False else: raise GadgetServerError( "Unexpected value for @@GLOBAL.GTID_MODE: {0}." "Expected: 'ON' or 'OFF'.".format(res[0][0]), server=self) def get_gtid_executed(self, skip_gtid_check=True): """Get the executed GTID set of the server. This function retrieves the (current) GTID_EXECUTED set of the server. :param skip_gtid_check: Flag indicating if the check for GTID support will be skipped or not. By default 'True' (check is skipped). :returns a string with the GTID_EXECUTED set for this server. :rtype str :raises GadgetError: if GTIDs are not supported or not enabled. """ if not skip_gtid_check: # Check server for GTID support. gtid_support = self.supports_gtid() if not gtid_support: raise GadgetServerError("Global Transaction IDs are not " "supported.", server=self) # Get GTID_EXECUTED. try: return self.exec_query("SELECT @@GLOBAL.GTID_EXECUTED")[0][0] except GadgetQueryError: if skip_gtid_check: # Query likely failed because GTIDs are not supported, # therefore skip error in this case. return "" else: # If GTID check is not skipped re-raise exception. raise except IndexError: # If no rows are returned by query then return an empty string. return "" def supports_plugin(self, plugin, state='ACTIVE'): """Check if the given plugin is supported. Check if the server supports the specified plugin. Return True if plugin is installed and active. :param plugin: Name of plugin to check :type plugin: string :param state: the expected plugin state to check, by default ACTIVE. :type state: string :returns: True if the plugin is supported and it has the given state, False otherwise. :rtype: boolean :raise GadgetServerError: If an error occurs when checking for the plugin support. """ _PLUGIN_QUERY = Query( "SELECT PLUGIN_NAME, PLUGIN_STATUS " "FROM INFORMATION_SCHEMA.PLUGINS WHERE PLUGIN_NAME LIKE ?", "{0}%".format(plugin)) res = self.exec_query(_PLUGIN_QUERY) if not res: # plugin not found. _LOGGER.debug("Plugin %s is not installed", plugin) return False elif res[0][1].upper() != state.upper(): # The state is not the desired. _LOGGER.debug("Plugin %s has state: %s and not the expected: %s", plugin, res[0][1], state) return False return True def get_all_databases(self, ignore_internal_dbs=True): """Get all databases from the server. Retrieve the list of all databases on the server, except for internal databases (e.g., INFORMATION_SCHEMA and PERFORMANCE_SCHEMA) if the 'ignore_internal_dbs' parameter is set to True. Note: New internal database 'sys' added by default for MySQL 5.7.7+. :param ignore_internal_dbs: Ignore internal databases. :type ignore_internal_dbs: boolean :returns: Result with the name of all the databases in the server. :rtype: list of tuples :raise GadgetServerError: If an error occurs when getting the list of all databases. """ if ignore_internal_dbs: _GET_DATABASES = """ SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME != 'INFORMATION_SCHEMA' AND SCHEMA_NAME != 'PERFORMANCE_SCHEMA' """ # Starting from MySQL 5.7.7, sys schema is installed by default. if self.check_version_compat(5, 7, 7): _GET_DATABASES = ("{0} AND SCHEMA_NAME != 'sys'" "".format(_GET_DATABASES)) else: _GET_DATABASES = """ SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA """ return self.exec_query(_GET_DATABASES) def read_and_exec_sql(self, input_file, verbose=False): """Read an input file containing SQL statements and execute them. :param input_file: The full path to the SQL file. :type input_file: string :param verbose: Log the read statements. By default, False. :type verbose: boolean :raise GadgetServerError: If an error occurs when executing the statements in the SQL file. """ with open(input_file) as f_input: while True: cmd = f_input.readline() if not cmd: break if len(cmd) > 1: if cmd[0] != '#': if verbose: _LOGGER.debug("%s", cmd) query_options = { 'fetch': False } self.exec_query(cmd, query_options) def binlog_enabled(self): """Check binary logging status. Check if binary logging is enabled on the server based on the value of the 'log_bin' variable. :return: False if binary logging is disabled and True otherwise. :rtype: boolean :raise GadgetServerError: If an error occur when trying to get the value of the 'log_bin' variable. """ try: res = self.show_server_variable("log_bin") except GadgetServerError as err: raise GadgetServerError("Cannot get value of 'log_bin' variable. " "{0}".format(err.errmsg), cause=err, server=self) if not res: raise GadgetServerError( "No value returned for 'log_bin' variable.", server=self) if res[0][1] in ("OFF", "0"): return False return True def toggle_global_read_lock(self, enable=True): """ Enable or disable read-only mode on the server. Note: user must have SUPER privilege :param enable: Indicate if read-only mode will be enabled (True) or disabled (False). If True (default) then flush all tables with read lock and set the [super_]read_only to 'ON'. If False then set [super_]read_only mode to 'OFF' and unlock all tables. :type enable: boolean """ # Starting with version 5.7.8 MySQL has a new super_read_only system # variable has_super_read_only = self.check_version_compat(5, 7, 8) var_name = "super_read_only" if has_super_read_only else "read_only" if enable: self.exec_query("FLUSH TABLES WITH READ LOCK") self.set_variable(var_name, "ON", "global") else: self.set_variable(var_name, "OFF", "global") self.exec_query("UNLOCK TABLES") def toggle_binlog(self, action='disable'): """Enable or disable binary logging. Note: User must have SUPER privilege. :param action: if 'disable' then turn off the binary logging. if 'enable' then turn on binary logging. If none of the previous action is specified then nothing is done (no action). :type action: 'disable' or 'enable' :raise GadgetServerError: If an error occur when setting the SQL_LOG_BIN value. """ if action.lower() == 'disable': self.exec_query("SET SQL_LOG_BIN=0") elif action.lower() == 'enable': self.exec_query("SET SQL_LOG_BIN=1") def get_server_id(self): """Retrieve the server ID. :return: Value of the 'server_id' variable. :rtype: integer :raise GadgetServerError: If an error occurs when trying to retrieve the server ID. """ try: res = self.show_server_variable("server_id") except GadgetServerError as err: raise GadgetServerError("Cannot retrieve 'server_id'. " "{0}".format(err.errmsg), cause=err, server=self) return int(res[0][1]) def get_server_uuid(self): """Retrieve the UUID of the server. :return: Value of the 'server_uuid' variable. If not available None is returned. :rtype: string :raises GadgetServerError: If an errors occurs when trying to retrieve the server UUID. """ try: res = self.show_server_variable("server_uuid") if res is None or res == []: return None except GadgetServerError as err: raise GadgetServerError("Cannot retrieve 'server_uuid': " "{0}".format(err.errmsg), cause=err, server=self) return res[0][1] def grant_tables_enabled(self): """Check if grant tables is enabled on the server. In other words, this functions checks if the privileges system is enabled on the server. Grant tables might be disabled if the --skip-grant-tables option is used when starting the server. :return: True if grant tables (privileges system) is enabled otherwise False (disabled). :rtype: boolean """ if self.grants_enabled is None: try: self.exec_query("SHOW GRANTS FOR 'snuffles'@'host'") self.grants_enabled = True except (GadgetCnxError, GadgetQueryError) as err: if (err.args[0] == ER_OPTION_PREVENTS_STATEMENT and "--skip-grant-tables" in err.errmsg): self.grants_enabled = False # Ignore other errors as they are not pertinent to the check else: self.grants_enabled = True return self.grants_enabled def get_server_binlogs_list(self, include_size=False): """Get the binary log file names listed on a server. Obtains the binlog file names available on the server by using the 'SHOW BINARY LOGS' statement, returning these file names as a list. :param include_size: Indicate if the returning list shall include the size of the file. :type include_size: boolean :return: List with the binary log file names available on the server. :rtype: list :raises GadgetServerError: If an errors occurs when executing SHOW BINARY LOGS. """ res = self.exec_query("SHOW BINARY LOGS") server_binlogs = [] for row in res: if include_size: server_binlogs.append(row) else: server_binlogs.append(row[0]) return server_binlogs def is_plugin_installed(self, plugin_name, is_active=False, silence_warnings=False): """Test if the given plugin is installed/loaded in this server. :param plugin_name: The name of the plugin to load. :type plugin_name: str. :param is_active: If True verifies the plugin is also in 'ACTIVE' state. :type is_active: bool :param silence_warnings: If True avoids logging warning messages. :type silence_warnings: bool (False by default) :return: True if the plugin is installed otherwise False. :rtype: bool. """ res = self.exec_query("show plugins") for row in res: if plugin_name in row[0]: if is_active: return True if "ACTIVE" in row[1] else False else: return True if not silence_warnings: _LOGGER.warning("The %s plugin has not been " "installed/loaded in %s", plugin_name, self) return False def install_plugin(self, plugin_name): """Attempts to install/load the given plugin in this server. :param plugin_name: The name of the plugin to load. :type plugin_name: str. :raise GadgetServerError: If the GR plugin could not be loaded. :return: True if the plugin is successfully loaded or was already loaded, else an exception is raised. :rtype: boolean :raise GadgetServerError: If the plugin could not be installed. """ msg = "Initializing {0} plugin on {1}".format(plugin_name, self) _LOGGER.info(msg) if self.is_plugin_installed(plugin_name, silence_warnings=True): return True else: try: if "WIN" in self.select_variable("version_compile_os").upper(): ext = ".dll" else: ext = ".so" self.exec_query("INSTALL PLUGIN {plugin_name} SONAME " "'{plugin_name}{ext}'" "".format(plugin_name=plugin_name, ext=ext)) _LOGGER.debug("The %s plugin has been successfully install " "in server: %s", plugin_name, self) except GadgetQueryError as err: if "already exists" in err.errmsg: _LOGGER.debug("The %s plugin is already installed: %s" "", plugin_name, err.errmsg) else: _LOGGER.error("An error was found trying to install the " "%s plugin: %s", plugin_name, err.errmsg) raise GadgetServerError("The {0} plugin could not be " "loaded in the server {1}" "".format(plugin_name, self), cause=err) return True def uninstall_plugin(self, plugin_name): """Attempts to uninstall the given plugin in this server. :param plugin_name: The name of the plugin to unload. :type plugin_name: str. :raise GadgetServerError: If the plugin could not be uninstalled. """ try: self.exec_query("UNINSTALL PLUGIN {plugin}" "".format(plugin=plugin_name)) except GadgetQueryError as err: if "does not exist" in err.errmsg: _LOGGER.debug("The %s plugin is not installed: %s" "", plugin_name, err.errmsg) else: _LOGGER.error("An error was found trying to uninstall the %s " "plugin: %s", plugin_name, err.errmsg) raise GadgetServerError("The {0} plugin could not be " "uninstalled in the server {1}" "".format(plugin_name, self), cause=err) def start_plugin(self, plugin_name): """Starts the given plugin :param plugin_name: The name of the plugin to load. :type plugin_name: str. """ self.exec_query("START {plugin}".format(plugin=plugin_name)) def stop_plugin(self, plugin_name): """Stops the given plugin :param plugin_name: The name of the plugin to load. :type plugin_name: str. """ self.exec_query("STOP {plugin}".format(plugin=plugin_name)) def __str__(self): """String representation of the class Server :return: representation the server with information of the host and port. :rtype: string """ if self.socket and os.name == "posix": return "'{0}:{1}'".format(self.host, self.socket) else: return "'{0}:{1}'".format(self.host, self.port) class QueryKillerThread(threading.Thread): """Class to run a thread to kill an executing query. This class is used to spawn a thread than will kill the execution (connection) of a query upon reaching a given timeout. """ def __init__(self, server, query, timeout): """Constructor. :param server: Server instance where the target query is executed. :type server: Server object :param query: Target query to kill. :type query: string :param timeout: Timeout value in seconds used to kill the query when reached. :type timeout: integer """ threading.Thread.__init__(self) self._stop_event = threading.Event() self._query = query self._timeout = timeout self._server = server self._connection = server.get_connection() server.get_version() def run(self): """Main execution of the query killer thread. Stop the thread if instructed as such """ connector_error = None # Kill the query connection upon reaching the given execution timeout. while not self._stop_event.is_set(): # Wait during the defined time. self._stop_event.wait(self._timeout) # If the thread was asked to stop during wait, it does not try to # kill the query. if not self._stop_event.is_set(): try: cur = None # Get process information from threads table when available # (for versions > 5.6.1), since it does not require a mutex # and has minimal impact on server performance. cur = MySQLUtilsCursorResult(self._connection.run_sql( "SELECT processlist_id " "FROM performance_schema.threads" " WHERE processlist_command='Query'" " AND processlist_info='{0}'".format(self._query))) result = cur.fetchall() try: process_id = result[0][0] except IndexError: # No rows are returned if the query ended in the # meantime. process_id = None # Kill the connection associated to que process id. # Note: killing the query will not work with # connector-python,since it will hang waiting for the # query to return. if process_id: self._connection.run_sql("KILL {0}".format(process_id)) except mysqlsh.DBError as err: # Hold error to raise at the end. connector_error = err finally: # Close cursor if available. if cur: cur.close() # Stop this thread. self.stop() # Close connection. try: self._connection.disconnect() except mysqlsh.DBError: # Only raise error if no previous error has occurred. if not connector_error: raise finally: # Raise any previous error that already occurred. if connector_error is not None: # pylint: disable=E0702 raise connector_error def stop(self): """Stop the thread. Set the event flag for the thread to stop as soon as possible. """ self._stop_event.set() class LocalErrorLog(object): """This class can be used to read the local Error log file. Note: In the case the log_error variable is set to "stderr", then it will not be possible to retrieve the messages. """ def __init__(self, server, raise_error=False): """Constructor Create the LocalErrorLog object setting the server to retrieve messages logged to its error log file. :param server: The server to retrieve messages from the error log. :type server: mysql_gadgets.common.server.Server :raise GadgetServerError: If the "log_error" var is set to stderr instead of a file. """ self.log_error = server.select_variable("log_error") if self.log_error == "stderr": if raise_error: raise GadgetServerError("The log error is set to stderr") self._log_file = None else: datadir = server.select_variable("datadir") self._log_file = get_abs_path(self.log_error, datadir) def get_size(self): """Get the current size of the log error file. Returns the current size of the server log error file. This method can be used as a starting point to read from the file. :return: Size of the error log file. :rtype: int """ if self._log_file: return os.stat(self._log_file).st_size return None def read(self, offset, errors_only=True): """Reads the Error log file. :param offset: Starting position. :type offset: int :param errors_only: If the messages returned should be "ERROR" messages only, By default True. :type errors_only: boolean :return: Messages logged at the logging file. :rtype: str """ if self._log_file: with open(self._log_file, "r") as e_file: if offset: e_file.seek(offset) result = [] for line in e_file: if errors_only and "[ERROR]" not in line: continue result.append(line) return "".join(result) return "" mysql_gadgets/common/logger.py 0000644 00000011645 15231164450 0012540 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 logging import sys import mysqlsh from mysql_gadgets.exceptions import GadgetLogError shell = mysqlsh.globals.shell PY2 = int(sys.version[0]) == 2 # Custom logging level. STEP_LOG_LEVEL_NAME = 'STEP' STEP_LOG_LEVEL_VALUE = 25 def setup_logging(verbosity=0): """Setup logging for mysqlprovision. This method setup the logging configuration (i.e., verbosity) and logging to be performed to the terminal and file through the shell log. :param verbosity: Verbosity level used to setup the logging level. If > 0 then DEBUG otherwise INFO. By default 0. :type verbosity: Integer """ # Set Logger class to use. logging.setLoggerClass(CustomLevelLogger) # Determine the logging level to use based on verbosity. # Note: Set at the root logger level to be used by default. logging_level = logging.DEBUG if verbosity > 1 else logging.INFO logging.getLogger().setLevel(logging_level) class CustomLevelLogger(logging.getLoggerClass()): """ Custom Logger Logs messages directly through shell. """ def __init__(self, name, level=logging.NOTSET): """ Constructor. :param name: Name of the logger. :type name: string :param level: Initial logging level for the logger. By default, NOTSET. :type level: string """ # Use old-class initialization for compatibility with Python 2.6 # and add logging.getLoggerClass() check to avoid recursion issues. # super(CustomLevelLogger, self).__init__(name, level) if logging.getLoggerClass() != CustomLevelLogger: logging.getLoggerClass().__init__(self, name, level) else: logging.Logger.__init__(self, name, level) logging.addLevelName(STEP_LOG_LEVEL_VALUE, STEP_LOG_LEVEL_NAME) def step(self, msg, *args, **kwargs): """Log message of STEP severity.""" self.log(STEP_LOG_LEVEL_VALUE, msg, *args, **kwargs) def debug(self, msg, *args, **kwargs): """Log message of DEBUG severity (function overwrite).""" self.log(logging.DEBUG, msg, *args, **kwargs) def info(self, msg, *args, **kwargs): """Log message of INFO severity (function overwrite).""" self.log(logging.INFO, msg, *args, **kwargs) def warning(self, msg, *args, **kwargs): """Log message of WARNING severity (function overwrite).""" self.log(logging.WARNING, msg, *args, **kwargs) def error(self, msg, *args, **kwargs): """Log message of ERROR severity (function overwrite).""" self.log(logging.ERROR, msg, *args, **kwargs) def critical(self, msg, *args, **kwargs): """Log message of CRITICAL severity (function overwrite).""" self.log(logging.CRITICAL, msg, *args, **kwargs) def log(self, level, msg, *args, **kwargs): """ Log message of any severity (function overwrite). """ if kwargs: raise GadgetLogError("Invalid use of logging parameters. kwargs " "not supported for log(): " "{0}".format(kwargs)) levellog = levelname = logging.getLevelName(level) if levelname == "CRITICAL": levellog = levelname = "ERROR" if levelname == STEP_LOG_LEVEL_NAME: levellog = "INFO" formatted_msg = msg % args if self.isEnabledFor(level): if level >= logging.ERROR: output = sys.stderr else: output = sys.stdout output.write( u"%s\n" % { "type": levelname, "msg": formatted_msg.encode('utf-8') if PY2 else formatted_msg }) shell.log(levellog, "mysqlprovision: " + formatted_msg) mysql_gadgets/common/connection_parser.py 0000644 00000046515 15231164450 0015000 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """ This module contains the methods to support common operations over the ip address or hostnames and parsing of connection strings. """ import re import os import logging # Use backported OrderedDict if not available (for Python 2.6) try: from collections import OrderedDict except ImportError: from ordered_dict_backport import OrderedDict from mysql_gadgets.common.logger import CustomLevelLogger from mysql_gadgets.exceptions import GadgetCnxFormatError logging.setLoggerClass(CustomLevelLogger) _LOGGER = logging.getLogger(__name__) _BAD_CONN_FORMAT = (u"Connection '{0}' cannot be parsed. Please review the " u"used connection string (accepted formats: " u"<user>@<host>[:<port>][:<socket>]") _BAD_QUOTED_HOST = u"Connection '{0}' has a malformed quoted host" _INVALID_PORT = (u"Invalid port '{0}' for connection string '{1}'. " u"Port must be >= 0 and <= 65535.") _UNPARSED_CONN_FORMAT = ("Connection '{0}' not parsed completely. Parsed " "elements '{1}', unparsed elements '{2}'") _CONN_QUOTEDHOST = re.compile( r"((?:^[\'].*[\'])|(?:^[\"].*[\"]))" # quoted host name r"(?:\:(\d+))?" # Optional port number r"(?:\:([\/\\w+.\w+.\-]+))?" # Optional path to socket ) _CONN_LOGINPATH = re.compile( r"((?:\\\"|[^:])+|(?:\\\'|[^:])+)" # login-path r"(?:\:(\d+))?" # Optional port number r"(?:\:([\/\\w+.\w+.\-]+))?" # Optional path to socket ) _CONN_CONFIGPATH = re.compile( r"([\w\:]+(?:\\\"|[^[])+|(?:\\\'|[^[])+)" # config-path r"(?:\[([^]]+))?", # group re.U ) _CONN_ANY_HOST = re.compile( r"""( (?![.-]) # Match cannot start by '.' or '-' [\w._-]* # start with 0 or more: alphanum or '%' % # must have a wildcard '%' at least [\w.%_-]* # then 0 or more: alphanum, '%', '.' or '-' (?<![.-]) # but cannot be ended by '.' or '-'. ) (?::(.+))?$ # capture all the rest """, re.VERBOSE) _CONN_HOST_NAME = re.compile( r"""( (?: (?: (?: (?!-) # must not start with hyphen '-' (?:[\w\d-])* # must not end with the hyphen [A-Za-z] # starts with a character from the alphabet (?:[\w\d-])* (?: (?<!-) # end capturing if a '-' is followed by '.' ) ){1,63} # limited length for segment ) (?: # following segments (?:\.)? # the segment separator the dot '.' (?: (?!-) [\w\d-]{1,63} # last segment (?<!-) #shuld not end with hyphen ) )* ) ) (.*) # capture all the rest """, re.VERBOSE) _CONN_IPV4_NUM_ONLY = re.compile( r"""( (?: # start of the IPv4 1st group 25[0-4] # this match numbers 250 to 254 | # or 2[0-4]\d # this match numbers from 200 to 249 | # or 1\d\d # this match numbers from 100 to 199 | # or [1-9]{0,1}\d # this match numbers from 0 to 99 ) (?: # start of the 3 next groups \. # the prefix '.' like in '.255' (?: 25[0-4]|2[0-4]\d|1\d\d|[1-9]?\d # same group as before ) ) {3} # but it will match 3 times of it and prefixed by '.' ) (?:\:{0,1}(.*)) """, re.VERBOSE) _CONN_PORT_ONLY = re.compile( r"""(?: \]{0,1} # the ']' of IPv6 -optional \:{0,1} # the ':' for port number ( -\d+|\d* # matches any sequence of numbers, ) # including negative/invalid port numbers ) # end of port number group (?:\:{0,1}(.*)) # all the rest to extract the socket """, re.VERBOSE) _CONN_SOCKET_ONLY = re.compile( r"""(?: # Not capturing group of ':' \:{0,1} ([ # Capturing '\' or '/' file name.ext \/\\w+.\w+.\- ]+ # to match a path ) )? (.*) # all the rest to advice the user. """, re.VERBOSE) _CONN_IPV6 = re.compile( r""" \[{0,1} # the optional heading '[' ( (?!.*::.*::) # Only a single whildcard allowed (?:(?!:)|:(?=:)) # Colon iff it would be part of a wildcard (?: # Repeat 6 times: [0-9a-f]{0,4} # A group of at most four hexadecimal digits (?:(?<=::)|(?<!::):) # Colon unless preceded by wildcard ){6} # expecting 6 groups (?: # Either [0-9a-f]{0,4} # Another group (?:(?<=::)|(?<!::):) # Colon unless preceded by wildcard [0-9a-f]{0,4} # Last group (?:(?<=::) # Colon iff preceded by exacly one colon |(?<!:) |(?<=:)(?<!::): ) ) ) (?: \]{0,1}\:{0,1}(.*) # optional closing ']' and group for the rest ) """, re.VERBOSE) # Type of address amd Key names for the dictionary IP_MATCHERS HN = "hostname" IPV4 = "IPv4" IPV6 = "IPv6" ANY_LIKE = "host like" # This dictionary is used to identify the matched type.. IP_MATCHERS = OrderedDict([ (IPV4, _CONN_IPV4_NUM_ONLY), (IPV6, _CONN_IPV6), (ANY_LIKE, _CONN_ANY_HOST), (HN, _CONN_HOST_NAME), ]) _DEFAULT_PORT = 3306 def hostname_is_ip(hostname): """Check if specified hostname is an IP address. :param hostname: hostname or IP address to check. :type hostname: string :return: True if the specified 'hostname' is a valid IP address (IPv4 or IPv6), otherwise False. :rtype: boolean """ if len(hostname.split(":")) <= 1: # if fewer colons, must be IPv4 grp = _CONN_IPV4_NUM_ONLY.match(hostname) else: grp = _CONN_IPV6.match(hostname) if not grp: return False return True def parse_connection(connection_str, options=None): """Parse connection values. The function parses a connection string with the following format: user@host[:port][:socket]. A dictionary is returned containing the connection parameters. The function is designed so that it shall be possible to use it with a ``connect`` call in the following manner: cnx_options = parse_connection(connection_values, options) cnx = mysql.connector.connect(**cnx_options) Notes: This method validates IPv4 addresses and standard IPv6 addresses. This method accepts quoted host portion strings. If the host is marked with quotes, the code extracts this without validation and assigns it to the host variable in the returned tuple. This allows users to specify host names and IP addresses that are outside of the supported validation. :param connection_str: Connection string in the form: user@host[:port][:socket]. :type connection_str: string :param options: Dictionary of options (e.g. charset, ssl_cert, ssl_ca, ssl_key, ssl). :type options: dictionary :return: dictionary with the parsed connection values (user, host, port, etc.). :rtype: dictionary :raises GadgetCnxFormatError: if a parsing error occurs. """ if options is None: options = {} # SSL options, must not be overwritten with those from options. ssl_ca = None ssl_cert = None ssl_key = None ssl = None # Split on the '@' to determine the connection string format. # The user/password may have the '@' character, split by last occurrence. conn_format = connection_str.rsplit('@', 1) if len(conn_format) == 2: # Handle as in the format: user@host[:port][:socket] user, hostportsock = conn_format # Handle host, port and socket if len(hostportsock) <= 0 or len(user) <= 0: raise GadgetCnxFormatError( _BAD_CONN_FORMAT.format(connection_str)) if hostportsock[0] in ('"', "'"): # Need to strip the quotes res = _match(_CONN_QUOTEDHOST, hostportsock) if res and len(res) == 3: host = res[0] port = res[1] socket = res[2] else: raise GadgetCnxFormatError( _BAD_CONN_FORMAT.format(connection_str)) if host[0] == '"': host = host.strip('"') if host and host[0] == "'": host = host.strip("'") # Raise exception if no host is specified, e.g. user@"" if not host: raise GadgetCnxFormatError( _BAD_CONN_FORMAT.format(connection_str)) else: host, port, socket, _ = parse_server_address(hostportsock) else: # Unrecognized format raise GadgetCnxFormatError(_BAD_CONN_FORMAT.format(connection_str)) # Get character-set from options if isinstance(options, dict): charset = options.get("charset", None) # If one SSL option was found before, not mix with those in options. if not ssl_cert and not ssl_ca and not ssl_key and not ssl: ssl_cert = options.get("ssl_cert", None) ssl_ca = options.get("ssl_ca", None) ssl_key = options.get("ssl_key", None) ssl = options.get("ssl", None) else: charset = None ssl_cert = None ssl_ca = None ssl_key = None ssl = None _LOGGER.warning("options is not instance of dict: %s", options) # Set parsed connection values connection = { "user": user, "host": host, } if charset: connection["charset"] = charset if ssl_cert: connection["ssl_cert"] = ssl_cert if ssl_ca: connection["ssl_ca"] = ssl_ca if ssl_key: connection["ssl_key"] = ssl_key if ssl: connection["ssl"] = ssl # If a port was specified we use the port, if a socket was specified # we use the socket, else if we specify neither, we use the # default port. if port: port = int(port) # Check if port range is valid. if port < 0 or port > 65535: raise GadgetCnxFormatError(_INVALID_PORT.format(port, connection_str)) connection["port"] = port elif socket is not None and os.name == "posix": connection['unix_socket'] = socket # Port is mandatory to create Server instance. connection["port"] = None else: connection["port"] = _DEFAULT_PORT return connection def parse_server_address(connection_str): """Parses host, port and socket from the given connection string. :param connection_str: Connection string in the form: user@host[:port][:socket]. :type connection_str: string :return: a tuple of (host, port, socket, add_type) where add_type is the name of the parser that successfully parsed the hostname from the connection string. :rtype: tuple :raises GadgetCnxFormatError: if a parsing error occurs. """ # Default values to return. host = None port = None socket = None address_type = None unparsed = None # From the matchers look the one that match a host. for ip_matcher in IP_MATCHERS: try: group = _match(IP_MATCHERS[ip_matcher], connection_str) if group: host = group[0] if ip_matcher == IPV6: host = "[%s]" % host if group[1]: part2_port_socket = _match(_CONN_PORT_ONLY, group[1], throw_error=False) if not part2_port_socket: unparsed = group[1] else: port = part2_port_socket[0] if part2_port_socket[1]: part4 = _match(_CONN_SOCKET_ONLY, part2_port_socket[1], throw_error=False) if not part4: unparsed = part2_port_socket[1] else: socket = part4[0] unparsed = part4[1] # If host is match we stop looking as is the most significant. if host: address_type = ip_matcher break # ignore the error trying to match. except GadgetCnxFormatError: pass # we must alert, that the connection could not be parsed. if host is None: raise GadgetCnxFormatError(_BAD_CONN_FORMAT.format(connection_str)) _verify_parsing(connection_str, host, port, socket, address_type, unparsed) _LOGGER.debug("->parse_server_address \n host: %s \n address_type: %s", host, address_type) return host, port, socket, address_type def _verify_parsing(connection_str, host, port, socket, address_type, unparsed): """Verify connection string parsing. Verify that the connection string was totally parsed and not parts of it where not matched, otherwise raise an error. :param connection_str: Connection string in the form: user@host[:port][:socket]. :type connection_str: string :param host: the parsed host :type host: string :param port: the parsed port :type port: string :param socket: the parsed socket :type socket: string :param address_type: Type of the parsed host, one of the following value: "IPv4", "IPv6" or "hostname" :type address_type: string :param unparsed: unparsed string (not identified part) :type unparsed: string :raises GadgetCnxFormatError: if a part of the connection string was not identified while parsed. """ exp_connection_str = connection_str # _LOGGER.debug("exp_connection_str %s", exp_connection_str) parsed_connection_list = [] if host: # _LOGGER.debug("host %s", host) if address_type == IPV6 and "[" not in connection_str: host = host.replace("[", "") host = host.replace("]", "") parsed_connection_list.append(host) if port: # _LOGGER.debug("port %s", port) parsed_connection_list.append(port) if socket: # _LOGGER.debug("socket %s", socket) parsed_connection_list.append(socket) parsed_connection = ":".join(parsed_connection_list) # _LOGGER.debug('parsed_connection %s', parsed_connection) diff = None if not unparsed: # _LOGGER.debug('not unparsed found, creating diff') diff = connection_str.replace(host, "") if port: diff = diff.replace(port, "") if socket: diff = diff.replace(socket, "") # _LOGGER.debug("diff %s", diff) # _LOGGER.debug("unparsed %s", unparsed) if unparsed or (exp_connection_str != parsed_connection and (diff and diff != ":")): # _LOGGER.debug("raising exception") parsed_args = "host:%s, port:%s, socket:%s" % (host, port, socket) err_msg = _UNPARSED_CONN_FORMAT.format(connection_str, parsed_args, unparsed) # _LOGGER.warning(err_msg) raise GadgetCnxFormatError(err_msg) def _match(pattern, connection_str, throw_error=True): """Check pattern match with connection string. Tries to match a pattern with the connection string and returns the groups. :param pattern: Regular expression object used to parse the connection string. :type pattern: re.RegexObject :param connection_str: Connection string in the form: user@host[:port][:socket]. :type connection_str: string :param throw_error: Indicate if an exception is raised (True) if the connection string does not match the pattern, or False is returned. By default: True. :type throw_error: boolean :return: tuple containing all the subgroups of the matched pattern. If no match is found False is returned if 'throw_error' is set to False. :rtype: tuple or boolean :raises GadgetCnxFormatError: if connection string does not match the given pattern and throw_error is set to True. """ grp = pattern.match(connection_str) if not grp: if throw_error: raise GadgetCnxFormatError(_BAD_CONN_FORMAT.format(connection_str)) return False return grp.groups() def clean_IPv6(host_address): """Clean IPv6 host address. :param host_address: host address (IPv6) :type host_address: string :return: the given host address without '[' and ']' characters (removed). :rtype: string """ if host_address: host_address = host_address.replace("[", "") host_address = host_address.replace("]", "") return host_address def parse_user_host(user_host): """Parse user, passwd, host, port from user:passwd@host :param user_host: MySQL user string (user:passwd@host) :type user_host: string :return: Tuple with the user and host. :rtype: tuple :raises: GadgetError if the user and host could not be parsed """ no_ticks = user_host.replace("'", "") try: conn_values = parse_connection(no_ticks) except GadgetCnxFormatError as err: raise GadgetCnxFormatError("Cannot parse user@host from: {0}." "".format(no_ticks), cause=err) return (conn_values['user'], conn_values['host']) mysql_gadgets/common/ordered_dict_backport.py 0000644 00000022134 15231164450 0015570 0 ustar 00 # # Copyright (c) 2017, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # # Backport of OrderedDict() class that runs on Python 2.6 # Based on the recipe suggested by the official Python documentation: # https://docs.python.org/2/library/collections.html#collections.OrderedDict. try: from thread import get_ident as _get_ident except ImportError: from dummy_thread import get_ident as _get_ident try: from _abcoll import KeysView, ValuesView, ItemsView except ImportError: pass class OrderedDict(dict): """Dictionary that remembers the order entries were added An inherited dict maps keys to values. The inherited dict provides __getitem__, __len__, __contains__, and get. The remaining methods are order-aware. The internal self.__map dictionary maps keys to links in a doubly linked list. The circular doubly linked list starts and ends with a sentinel element. The sentinel element never gets deleted (simplifying the algorithm). Each link is stored as a list of length three: [PREV, NEXT, KEY]. """ def __init__(self, *args, **kwds): """ Constructor Initialize an ordered dictionary. Signature is the same as for regular dictionaries, but keyword arguments are not recommended because their insertion order is arbitrary. """ if len(args) > 1: raise TypeError('expected at most 1 arguments, got %d' % len(args)) try: self.__root except AttributeError: self.__root = root = [] # sentinel node root[:] = [root, root, None] self.__map = {} self.__update(*args, **kwds) def __setitem__(self, key, value, dict_setitem=dict.__setitem__): """ Implements assignment to ordered_dict[key]: ordered_dict[key] = value Setting a new item creates a new link which goes at the end of the linked list, and the inherited dictionary is updated with the new key/value pair. """ if key not in self: root = self.__root last = root[0] last[1] = root[0] = self.__map[key] = [last, root, key] dict_setitem(self, key, value) def __delitem__(self, key, dict_delitem=dict.__delitem__): """Implements deletion of ordered_dict[key]: del ordered_dict[key] Deleting an existing item uses self.__map to find the link which is then removed by updating the links in the predecessor and successor nodes. """ dict_delitem(self, key) link_prev, link_next, key = self.__map.pop(key) link_prev[1] = link_next link_next[0] = link_prev def __iter__(self): """ Returns an iterator object during an iteration. """ root = self.__root curr = root[1] while curr is not root: yield curr[2] curr = curr[1] def __reversed__(self): """ Returns an iterator object during a reverse iteration. """ root = self.__root curr = root[0] while curr is not root: yield curr[2] curr = curr[0] def clear(self): """Remove all items from the dictionary. """ try: for node in self.__map.itervalues(): del node[:] root = self.__root root[:] = [root, root, None] self.__map.clear() except AttributeError: pass dict.clear(self) def popitem(self, last=True): """Remove and return a (key, value) pair from the dictionary. Pairs are returned in LIFO order if 'last' is true (default) or FIFO order if false. """ if not self: raise KeyError('dictionary is empty') root = self.__root if last: link = root[0] link_prev = link[0] link_prev[1] = root root[0] = link_prev else: link = root[1] link_next = link[1] root[1] = link_next link_next[0] = root key = link[2] del self.__map[key] value = dict.pop(self, key) return key, value def keys(self): """ Return a list with a copy of the dictionary keys. """ return list(self) def values(self): """ Return a list with a copy of the dictionary values. """ return [self[key] for key in self] def items(self): """ Return a list with a copy of the dictionary (key, value) pairs. """ return [(key, self[key]) for key in self] def iterkeys(self): """ Return an iterator over the dictionary keys. """ return iter(self) def itervalues(self): """ Return an iterator over the dictionary values. """ for k in self: yield self[k] def iteritems(self): """ Return an iterator over the dictionary (key, value) pairs. """ for k in self: yield (k, self[k]) def update(*args, **kwds): """ Update the dictionary with the key/value pairs from other. Update ordered_dict from dict/iterable D and I. If D is a dict instance, then: for k in D: ordered_dict[k] = D[k] If D has a keys() method, then: for k in D.keys(): ordered_dict[k] = D[k] Or if D is an iterable of items, then: for k, v in D: ordered_dict[k] = v In either case, this is followed by: for k, v in I.items(): ordered_dict[k] = v """ if len(args) > 2: raise TypeError('update() takes at most 2 positional ' 'arguments (%d given)' % (len(args),)) elif not args: raise TypeError('update() takes at least 1 argument (0 given)') self = args[0] # Make progressively weaker assumptions about "other" other = () if len(args) == 2: other = args[1] if isinstance(other, dict): for key in other: self[key] = other[key] elif hasattr(other, 'keys'): for key in other.keys(): self[key] = other[key] else: for key, value in other: self[key] = value for key, value in kwds.items(): self[key] = value # let subclasses override update without breaking __init__ __update = update __marker = object() def pop(self, key, default=__marker): """ Remove the key from the dictionary and return its value. If the key does not exist the default is returned (if specified). Remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised. """ if key in self: result = self[key] del self[key] return result if default is self.__marker: raise KeyError(key) return default def __repr__(self, _repr_running={}): """ Compute the "official" string representation of the dictionary. Used by repr(). In this case also used by str() and print since __str__() is not implemented. """ call_key = id(self), _get_ident() if call_key in _repr_running: return '...' _repr_running[call_key] = 1 try: if not self: return '%s()' % (self.__class__.__name__,) return '%s(%r)' % (self.__class__.__name__, self.items()) finally: del _repr_running[call_key] def __eq__(self, other): """ Used for equal comparison: ordered_dict == other Comparison to another OrderedDict is order-sensitive while comparison to a regular mapping is order-insensitive. """ if isinstance(other, OrderedDict): return len(self)==len(other) and self.items() == other.items() return dict.__eq__(self, other) def __ne__(self, other): """ Used for not equal comparison: ordered_dict <> other """ return not self == other mysql_gadgets/common/config_parser.py 0000644 00000151422 15231164450 0014100 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """ Module to manage (read and write) MySQL option files. """ from __future__ import print_function import codecs import logging import os import re import shutil import sys import stat # Use backported OrderedDict if not available (for Python 2.6) try: from collections import OrderedDict except ImportError: from ordered_dict_backport import OrderedDict from mysql_gadgets.exceptions import GadgetConfigParserError, GadgetError from mysql_gadgets.common.tools import get_abs_path, fs_encode, fs_decode from mysql_gadgets.common.logger import CustomLevelLogger # Get logger (must set class to custom logger to be used). logging.setLoggerClass(CustomLevelLogger) _LOGGER = logging.getLogger(__name__) PY2 = int(sys.version[0]) == 2 # pylint: disable=F0401 # pylint: disable=E0012, C0413, C0411 if PY2: from ConfigParser import (RawConfigParser, NoOptionError, NoSectionError, Error, MissingSectionHeaderError, ParsingError) else: from configparser import (RawConfigParser, NoOptionError, NoSectionError, Error, MissingSectionHeaderError, DuplicateSectionError, SectionProxy, DuplicateOptionError) unicode = str DEFAULT_EXTENSIONS = { 'nt': ('ini', 'cnf'), 'posix': ('cnf',) } MYSQL_OPTCRE_NV = re.compile( r'\s*' # any starting space/tab r'(?P<option>[^=]+?)' # option (permissive like original but not greedy) r'\s*(?:' # any number of space/tab, optionally followed by r'(?P<vi>[=])\s*' # separator (=) and any spaces and r'(?P<value>((?P<quote>[\'"]).*?(?P=quote))|' # value with quotes or r'([^#]*?)))?' # value (all except '#'), r'\s*(?:' # any number of space/tab, optionally followed by r'(?P<comment>#.*))?' # comment starting with '#', r'$' # up to eol ) # MySQL programs read startup options from the following files, in the # specified order. # See: http://dev.mysql.com/doc/refman/5.7/en/option-files.html DEFAULT_LOCATIONS = { 'nt': (r"%PROGRAMDATA%\MySQL\MySQL Server 5.7\my.ini", r"%PROGRAMDATA%\MySQL\MySQL Server 5.7\my.cnf", r"%WINDIR%\my.ini", r"%WINDIR%\my.cnf", r"C:\my.cnf", r"C:\my.ini"), 'posix': ("/etc/my.cnf", "/etc/mysql/my.cnf", "$MYSQL_HOME/my.cnf", "~/.my.cnf") } def option_list_to_dictionary(opt_list): """Converts a list of options to a dictionary. This function converts a list of option strings to a dictionary of options and respective values. E.g: ["port=13001", "binlog_format=ROW"] is transformed into {"port": "13001", "binlog_format": "ROW"} :param opt_list: list of option strings :type opt_list: list :return: dictionary with options and respective values. :rtype: dict Note: it converts hyphen '-' to underscores '_' on the value part. """ res = {} for opt in opt_list: opt_val = opt.split("=", 1) opt_name = opt_val[0].strip().replace("-", "_") try: val = opt_val[1].strip() except IndexError: # option without value val = None res[opt_name] = val return res def create_option_file(section_dict, name, prefix_dir=None, replace=False): """ Create an option file from a dictionary of dictionaries. :param section_dict: dictionary of dictionaries. The keys in the top level dictionary are sections and the values are dictionaries whose keys and values are the key:value pairs of the section. :type section_dict: {section1: {key: val, key2: val}, section2: {key: val..} ..} :param name: name of the config file. :type name: str :param prefix_dir: full path to a directory where we want the temporary file to be created. By default it uses the $HOME of the user. :type prefix_dir: str :param replace: if True, try to replace the file, if it already exists, otherwise throw an error saying the file already exists :type replace: bool :return: string with full path to the created config file. :rtype: str """ if prefix_dir is None: prefix_dir = os.path.expanduser("~") else: # check if prefix points to a valid folder # normalize path and expand possible ~ prefix_dir = os.path.normpath(os.path.expanduser(prefix_dir)) enc_prefix_dir = fs_encode(prefix_dir) if not os.path.isdir(enc_prefix_dir): raise GadgetError(u"prefix_dir '{0}' is not a valid folder. Check " u"if it exists.".format(prefix_dir)) _LOGGER.debug(u"Creating option file under directory %s ...", prefix_dir) f_path = os.path.join(prefix_dir, name) enc_f_path = fs_encode(f_path) # throw error if file exists and you don't want to replace it if os.path.exists(enc_f_path) and not replace: raise GadgetError(u"Unable to create option file '{0}' since a " u"file of the same already exists.".format(f_path)) try: f_handler = os.open(enc_f_path, os.O_CREAT | os.O_WRONLY, 0o600) except (OSError, IOError) as err: raise GadgetError(u"Unable to create option file '{0}': {1}." u"".format(f_path, unicode(err))) if f_handler: os.close(f_handler) _LOGGER.debug("Config file %s created successfully ", f_path) # Create configuration file if section_dict is not None: config = MySQLOptionsParser(f_path) _LOGGER.debug("Filling config parser object...") # Fill it with contents from options for section, section_d in section_dict.items(): config.add_section(section) for key, val in section_d.items(): config.set(section, key, val) _LOGGER.debug("Config parser object created.") _LOGGER.debug("Writing contents of the configuration file") config.write() _LOGGER.debug("Config file %s successfully written.", f_path) return f_path class MySQLOptionsParser(object): # pylint: disable=R0901 """This class implements methods to parse MySQL option files. Some properties for MySQL configuration files are different from the ones assumed by the Python implementation for the ConfigParser (Python 2) and configparser (Python 3) modules, requiring a custom implementation to handle them. Example of different properties: - Section (group) names are not case sensitive; - option names are case sensitive (like options in command line); - Only the '=' character is used to separate options with values; - By default, options without value are supported; - Only the '#' comment can start in the middle of a line; - Values can be optionally enclose within single or double quotation marks, which is useful if the value contains a '#' comment character; - specific escape sequences are supported; - there is no special 'default' section with precedence over other sections; - Specific directives are supported (!include and !includedir) to include configurations from other files. For more information, see: http://dev.mysql.com/doc/refman/5.7/en/option-files.html """ # pylint: disable=R0901 class MySQLRawConfigParser(RawConfigParser): """ This class customize the behaviour of the Python RawConfigParser. The default behaviour of the classes in the Python configparser module does not match the one used by MySQL to parse configuration files, requiring some properties and methods to be overwritten. For example, multiple line values are not supported for MySQL options files and should be disabled. """ def __init__(self, **kargs): # pylint: disable=E1002 """Constructor. """ # Call constructor of base class. if isinstance(RawConfigParser, type): # New style class super(MySQLOptionsParser.MySQLRawConfigParser, self).__init__( **kargs) else: # Old style class RawConfigParser.__init__(self, **kargs) # Option names are case sensitive for MySQL and you can use '_' or # '-' interchangeably self.optionxform = lambda option: option.replace('-', '_') # Set regexp used to parse options to use only '=' as valid # separator and ignore space at the start of an option. self._optcre = MYSQL_OPTCRE_NV if PY2: # Overwrite _read() for Python 2 to remove multiline support. def _read(self, fp, fpname): """Copy of base class method without multiline value code. Note: other minor adjustments were also made to meet the coding standard and avoid pylint issues. """ cursect = None # None, or a dictionary optname = None lineno = 0 e = None # None, or an exception while True: line = fp.readline() if not line: break lineno += 1 # comment or blank line? if line.strip() == '' or line[0] in '#;': continue if (line.split(None, 1)[0].lower() == 'rem' and line[0] in "rR"): # no leading whitespace continue # HERE: continuation line code (if ...) removed. # a section header or option header? else: # is it a section header? mo = self.SECTCRE.match(line) if mo: sectname = mo.group('header') if sectname in self._sections: cursect = self._sections[sectname] # HERE, remove default section special handling. else: cursect = self._dict() cursect['__name__'] = sectname self._sections[sectname] = cursect # So sections can't start with a continuation line optname = None # no section header in the file? elif cursect is None: raise MissingSectionHeaderError(fpname, lineno, line) # an option line? else: mo = self._optcre.match(line) if mo: optname, vi, optval = mo.group('option', 'vi', 'value') optname = self.optionxform(optname.rstrip()) # This check is fine because the OPTCRE cannot # match if it would set optval to None if optval is not None: if vi in ('=', ':') and ';' in optval: # ';' is a comment delimiter only if # it follows a spacing character pos = optval.find(';') if (pos != -1 and optval[pos - 1].isspace()): optval = optval[:pos] optval = optval.strip() # allow empty values if optval == '""': optval = '' # pylint: disable=E0012,E1136 cursect[optname] = [optval] else: # valueless option handling if cursect is not None: # pylint: disable=E0012,E1136 cursect[optname] = optval else: # a non-fatal parsing error occurred. set up # the exception but keep going. the exception # will be raised at the end of the file and # will contain a list of all bogus lines if not e: e = ParsingError(fpname) e.append(lineno, repr(line)) # if any parsing errors occurred, raise an exception if e: # pylint: disable=E0702 raise e # join the multi-line values collected while reading all_sections = [self._defaults] all_sections.extend(self._sections.values()) for options in all_sections: for name, val in options.items(): if isinstance(val, list): options[name] = '\n'.join(val) # Override add_section method to remove hardcoded default_section # value def add_section(self, section): """Create a new section in the configuration. Raise DuplicateSectionError if a section by the specified name already exists. Raise ValueError if name is DEFAULT or any of it's case-insensitive variants. """ # If condition from base class removed because we want to # allow a section with the name 'default' to be added. if section in self._sections: raise DuplicateSectionError(section) self._sections[section] = self._dict() else: # Overwrite _read() for Python 3 to remove multiline support. def _read(self, fp, fpname): """Copy of base class method without multiline value code. Note: other minor adjustments were also made to meet the coding standard and avoid pylint issues. """ elements_added = set() cursect = None # None, or a dictionary sectname = None optname = None lineno = 0 e = None # None, or an exception for lineno, line in enumerate(fp, start=1): comment_start = sys.maxsize # strip inline comments # pylint: disable=E1101 # Use dict comprehension syntax compatible with Python 2.6: # inline_prefixes = {p: -1 for p in # self._inline_comment_prefixes} inline_prefixes = self._inline_comment_prefixes if hasattr(self, '_inline_comment_prefixes') else self._prefixes.inline inline_prefixes = dict((p, -1) for p in inline_prefixes) while comment_start == sys.maxsize and inline_prefixes: next_prefixes = {} for prefix, index in inline_prefixes.items(): index = line.find(prefix, index + 1) if index == -1: continue next_prefixes[prefix] = index if index == 0 or ( index > 0 and line[index - 1].isspace()): comment_start = min(comment_start, index) inline_prefixes = next_prefixes # strip full line comments comment_prefixes = self._comment_prefixes if hasattr(self, '_comment_prefixes') else self._prefixes.full for prefix in comment_prefixes: if line.strip().startswith(prefix): comment_start = 0 break if comment_start == sys.maxsize: comment_start = None value = line[:comment_start].strip() if not value: if self._empty_lines_in_values: # add empty line to the value, but only if there # was no comment on the line # pylint: disable=E0012,E1136 if (comment_start is None and cursect is not None and optname and cursect[optname] is not None): cursect[optname].append( '') # newlines added at join continue # continuation line? # HERE: continuation line code (if ... else) removed. # a section header or option header? # is it a section header? mo = self.SECTCRE.match(value) if mo: sectname = mo.group('header') if sectname in self._sections: if self._strict and sectname in elements_added: raise DuplicateSectionError(sectname, fpname, lineno) cursect = self._sections[sectname] elements_added.add(sectname) elif sectname == self.default_section: cursect = self._defaults else: cursect = self._dict() self._sections[sectname] = cursect self._proxies[sectname] = SectionProxy(self, sectname) elements_added.add(sectname) # So sections can't start with a continuation line optname = None # no section header in the file? elif cursect is None: raise MissingSectionHeaderError(fpname, lineno, line) # an option line? else: mo = self._optcre.match(value) if mo: optname, _, optval = mo.group('option', 'vi', 'value') if not optname: e = self._handle_error(e, fpname, lineno, line) optname = self.optionxform(optname.rstrip()) if (self._strict and (sectname, optname) in elements_added): raise DuplicateOptionError(sectname, optname, fpname, lineno) elements_added.add((sectname, optname)) # This check is fine because the OPTCRE cannot # match if it would set optval to None if optval is not None: optval = optval.strip() # pylint: disable=E0012,E1136 cursect[optname] = [optval] else: # valueless option handling if cursect is not None: # pylint: disable=E0012,E1136 cursect[optname] = None else: # a non-fatal parsing error occurred. set up the # exception but keep going. the exception will be # raised at the end of the file and will contain a # list of all bogus lines e = self._handle_error(e, fpname, lineno, line) # if any parsing errors occurred, raise an exception if e: raise e # pylint: disable=E0702 # pylint: disable=E1101 self._join_multiline_values() # pylint: disable=E0203, W0201 def convert_sections_to_lower(self): """ Convert all sections to lower cases. Note: Sections are case insensitive, therefore all converted to lower cases. :raises GadgetConfigParserError: If a duplicated section is found when converting to lower case. """ # Convert all section names to lower case. new_sections = self._dict() for k in self._sections: v = self._sections[k] new_k = k.lower() if new_k in new_sections: raise GadgetConfigParserError( "File format error: section '{0}' is duplicated " "(sections are case insensitive).".format(k)) new_sections[new_k] = v self._sections = new_sections # Value for _proxies needs to be converted, only used for Python 3. if hasattr(self, "_proxies"): new_proxies = self._dict() for k in self._proxies: v = self._proxies[k] new_k = k.lower() new_proxies[new_k] = v self._proxies = new_proxies def __init__(self, filename, output_cnf_file=None): # pylint: disable=W0231 """Constructor. :param filename: filename of the option file to read :type filename: string """ if PY2: # NOTE: 'allow_no_value' parameter is not used due to compatibility # with Python 2.6. However, we set self._optcre with a regexp that # allow no value. kwargs = {} # Monkey patch ConfigParser DEFAULTSECT value, in order to handle # the 'default' section like any other (not as a special one). import ConfigParser ConfigParser.DEFAULTSECT = '' else: kwargs = {'allow_no_value': True, 'delimiters': ('=',), 'strict': False, 'empty_lines_in_values': False, 'interpolation': None, 'default_section': ''} # ConfigParser with information about configuration files included # from the configuration file specified as argument self._included_opt_parser = MySQLOptionsParser.MySQLRawConfigParser( **kwargs) # ConfigParser with information about the configuration file # specified as argument self._main_opt_parser = MySQLOptionsParser.MySQLRawConfigParser( **kwargs) self.default_extension = DEFAULT_EXTENSIONS[os.name] # get the absolute path for the provided filename self.filename = os.path.normpath(get_abs_path(filename, os.getcwd())) self.output_option_filename = output_cnf_file if output_cnf_file is not None: self.output_option_filename = os.path.normpath( get_abs_path(output_cnf_file, os.getcwd())) self._parse_option_file(self.filename) # Convert all section names to lower case. self._main_opt_parser.convert_sections_to_lower() self._included_opt_parser.convert_sections_to_lower() # Flag that must be set if any changes were made to the config read # from the file self.modified = False def _parse_option_file(self, filename): """Parse the given options file. This method parses a valid MySQL option file. It supports !include and !includedir directives and also parses all files included by those directives. Sor more information, see: http://dev.mysql.com/doc/refman/5.7/en/option-files.html :param filename: Absolute path to the option file to parse. :type filename: str :raises GadgetConfigParserError: If the given file or any of the included files is not readable. """ # Get files that need to be read, based on include directives. err_msg = "Option file '{0}' being included again in file '{1}'" files = [filename] for index, file_ in enumerate(files): try: enc_file_ = fs_encode(file_) with open(enc_file_, 'r') as op_file: for line in op_file: line = fs_decode(line) if line.startswith('!includedir'): _, dir_path = line.split(None, 1) dir_path = dir_path.strip() dir_path = get_abs_path(dir_path, file_) for entry in os.listdir(dir_path): entry = os.path.join(dir_path, entry) if (os.path.isfile(entry) and entry.endswith( self.default_extension)): # Only process files with valid extension. if entry in files: raise GadgetConfigParserError( err_msg.format(entry, file_)) files.insert(len(files), entry) else: # Skip all other files or directories. continue elif line.startswith('!include'): _, filename = line.split(None, 1) filename = filename.strip() filename = get_abs_path(filename, file_) if filename in files: raise GadgetConfigParserError(err_msg.format( filename, file_)) files.insert(len(files), filename) except (IOError, OSError) as err: raise GadgetConfigParserError( u"Unable to open option file '{0}': {1}" u"".format(self.filename, unicode(err)), cause=err) # Read configurations from option files. parse_err = u"File '{0}' could not be parsed: {1}" for index, file_ in enumerate(files): enc_file_ = fs_encode(file_) if index == 0: # main file is read to main parser only try: if PY2: if os.name == "nt": self._main_opt_parser.readfp( codecs.open(enc_file_, 'r', 'mbcs')) else: self._main_opt_parser.readfp( codecs.open(enc_file_, 'r', 'utf-8')) else: if os.name == "nt": self._main_opt_parser.read_file( codecs.open(enc_file_, 'r', 'mbcs')) else: self._main_opt_parser.read_file( codecs.open(enc_file_, 'r', 'utf-8')) except Error as err: raise GadgetConfigParserError( parse_err.format(file_, unicode(err)), cause=err) else: # config files from !include or !includedir are only added to # the include parser try: if PY2: if os.name == "nt": self._included_opt_parser.readfp( codecs.open(enc_file_, 'r', 'mbcs')) else: self._included_opt_parser.readfp( codecs.open(enc_file_, 'r', 'utf-8')) else: if os.name == "nt": self._included_opt_parser.read_file( codecs.open(enc_file_, 'r', 'mbcs')) else: self._included_opt_parser.read_file( codecs.open(enc_file_, 'r', 'utf-8')) except Error as err: raise GadgetConfigParserError( parse_err.format(enc_file_, unicode(err)), cause=err) def sections(self): """Return a list of the sections available. :return: List of available options. :rtype: list """ res = self._main_opt_parser.sections() + \ self._included_opt_parser.sections() # Remove duplicates but preserve the order seen = set() return [x for x in res if not (x in seen or seen.add(x))] def add_section(self, section): """Add a new section (group) for the main option file. :param section: Name of the section (group) to add. :type section: str :raises GadgetConfigParserError: If the section already exists on the main option file. """ # None or '' are not valid section names. if not section: raise GadgetConfigParserError("Cannot add an empty section.") # Sections are case insensitive (convert to lower case). section = section.lower() # If main config file has the section then raise an exception if self._main_opt_parser.has_section(section): raise GadgetConfigParserError( "Section '{0}' already exists.".format(section)) else: # Create it self._main_opt_parser.add_section(section) self.modified = True def has_section(self, section): """Indicate if the section (group) exists. This checks if the section exists in all read configuration, even for those read from files included with the !include or !includedir directives. :param section: Name of the section to check. :type section: str :return: True if the section exists False otherwise. :rtype: bool """ # Sections are case insensitive (convert to lower case). section = section.lower() return self._included_opt_parser.has_section(section) or \ self._main_opt_parser.has_section(section) def options(self, section): """Returns the list of options in the specified section (group). :param section: Name of the section (group) to get the options. :type section: str :return: All options available in the specified section. :rtype: list :raises GadgetConfigParserError: If the specified section does not exist. """ # Sections are case insensitive (convert to lower case). section = section.lower() res = [] no_section_included = False no_section_main = False try: res.extend(self._main_opt_parser.options(section)) except NoSectionError: no_section_main = True try: res.extend(self._included_opt_parser.options(section)) except NoSectionError: no_section_included = True # Raise error if the section is not found. # Note: not finding a section is different from finding it empty # (without any options). if no_section_included and no_section_main: raise GadgetConfigParserError("No section '{0}'.".format(section)) # Remove duplicates but preserve the order seen = set() return [x for x in res if not (x in seen or seen.add(x))] def has_option(self, section, option): """Indicate if the given option exists in the section (group). :param section: Name of the section (group) for the option to check. :type section: str :param option: Name of the option to check. :type option: str :return: True if the section (group) exists and contains the given option, False otherwise. :rtype: bool """ # Sections are case insensitive (convert to lower case). section = section.lower() return self._included_opt_parser.has_option(section, option) or \ self._main_opt_parser.has_option(section, option) def get(self, section, option): """Get the value for the given option and section. :param section: Name of the section (group). :type section: str :param option: Name of the option to get. :type option: str :return: Value of the option :rtype: str :raises GadgetConfigParserError: If the section does not exists or it does not contain the specified option. """ # Sections are case insensitive (convert to lower case). section = section.lower() # Since included files have higher precedence, first retrieve the # value from the include_parser and if not found then try to # retrieve it from the main file. try: val = self._included_opt_parser.get(section, option) except (NoOptionError, NoSectionError): pass else: return val # Option not found on the included files, now try main file. try: val = self._main_opt_parser.get(section, option) except NoSectionError: raise GadgetConfigParserError( "No section '{0}'.".format(section)) except NoOptionError: raise GadgetConfigParserError( "No option '{0}' in section '{1}'.".format(option, section)) else: return val def items(self, section): """Return a list of (name, value) pairs for each option in the section. :param section: Name of the section (group). :type section: str :return: All pairs (name, value) in the specified section, one for each option. :rtype: list(tuple(name, value)) :raises GadgetConfigParserError: If the specified section does not exist. """ # Sections are case insensitive (convert to lower case). section = section.lower() # First add items from main file no_section_main = False try: main_items = self._main_opt_parser.items(section) except NoSectionError: no_section_main = True main_items = tuple() no_section_include = False try: include_items = self._included_opt_parser.items(section) except NoSectionError: no_section_include = True include_items = tuple() # Raise error if the section is not found. # Note: section not found is different from empty section. if no_section_main and no_section_include: raise GadgetConfigParserError("No section '{0}'.".format(section)) res = OrderedDict(main_items) res.update(include_items) # List created by comprehension for compatibility with Py2 and Py3. return [(k, v) for k, v in res.items()] def set(self, section, option, value=None): """Set the option for the given section with the specified value. If the given section exists then set the given option to the specified value, otherwise raise GadgetConfigParserError. If no value is provided, the option is transformed into a valueless option. Important note: Only options in the main options file and not in the included files can be modified. :param section: Name of the section (group) for the option to set. :type section: str :param option: Name of the option to set. :type option: str :type value: Value to set for the given option and section. By default, None (valueless option added). :type value: str or None :raises GadgetConfigParserError: If the section does not exist, or exists only on an included option file, or the option exist in an included option file. """ # Sections are case insensitive (convert to lower case). section = section.lower() # If option exists on an included config file then raise error if self._included_opt_parser.has_option(section, option): raise GadgetConfigParserError( "Option '{0}' of section '{1}' cannot be set since " "it belongs to an included option file.".format(option, section)) # If section is only present on included files then raise error elif self._included_opt_parser.has_section(section) and not \ self._main_opt_parser.has_section(section): raise GadgetConfigParserError( "Section '{0}' cannot be modified since it belongs to an " "included option file.".format(section)) # If option not on included files (nor section only) then try to set it else: try: self._main_opt_parser.set(section, option, value) self.modified = True except NoSectionError: raise GadgetConfigParserError( "No section '{0}'.".format(section)) def remove_option(self, section, option): """Remove the specified option from the specified section. If the section does not exist, raise GadgetConfigParserError. If the option existed to be removed, return True; otherwise return False. Important note: Only options in the main option file and not in the included files can be removed. :param section: Name of the section (group) for the option to be removed. :type section: str :param option: Name of the option to remove. :type option: str :return: True if option existed and was removed else False :rtype: bool :raises GadgetConfigParserError: If the section does not exist or the option exists in an included option file. """ # Sections are case insensitive (convert to lower case). section = section.lower() # If option exists on an included config file then raise error if self._included_opt_parser.has_option(section, option): raise GadgetConfigParserError( "Option '{0}' of section '{1}' cannot be removed since " "it belongs to an included option file.".format(option, section)) # If option not on included files then try to remove it else: try: res = self._main_opt_parser.remove_option(section, option) if res: self.modified = True return res except NoSectionError: raise GadgetConfigParserError( "No section '{0}'.".format(section)) def remove_section(self, section): """Remove the specified section from the configuration. If the section in fact existed, return True. Otherwise return False. Important note: Only sections in the main option file and not in the included files can be removed. :param section: Name of the section (group) for the option to be removed. :type section: str :return: True if section existed as was removed otherwise False :rtype: bool :raises GadgetConfigParserError: If the section exists in an included option file. """ # Sections are case insensitive (convert to lower case). section = section.lower() # If option exists on an included config file then raise error if self._included_opt_parser.has_section(section): raise GadgetConfigParserError( "Section '{0}' cannot be removed since it belongs to an " "included option file.".format(section)) # If option not on included files (nor section only) then try to set it else: res = self._main_opt_parser.remove_section(section) if res: self.modified = True return res # pylint: disable=W0212 def write(self, backup_file_path=None): """Write configurations to the option file. If configurations were modified it saves those changes to file, the order of the sections/options is preserved as well as existing comments (except inline comments of deleted options or sections). :param backup_file_path: if provided, we try to create a create a backup of the original configuration file on the provided path. It must be an absolute path. :raises GadgetConfigParserError: If the user does not have permissions to overwrite the option file or to create a backup file (if asked to) or if an error occurred while parsing the configuration file. """ in_memory_sections = set(self._main_opt_parser.sections()) read_sections = set() read_options = set() drop_section = False cursect = None line = "" output_file = self.filename def optval_tostr(opt, val): """Auxiliary function used to obtain a formatted string with an option and option value. """ if val: return u"{0} = {1}".format(opt, val) return opt if not self.modified and self.output_option_filename is None: # if we did not do any modifications to what was read from file, # we can simply exit. return try: with open(fs_encode(self.filename), 'r') as f: lines = f.readlines() except IOError as err: raise GadgetConfigParserError( u"Option file '{0}' is not readable." u"".format(self.filename), cause=err) # Create a backup file if provided but no output_option file was # specified and if original file was not empty if (backup_file_path and os.stat(self.filename).st_size != 0 and not self.output_option_filename): if not os.path.isabs(backup_file_path): raise GadgetConfigParserError( u"'{0}' is not an absolute path. Please provide an " u"absolute path to the backup file".format( backup_file_path)) else: orig_perms = stat.S_IMODE(os.stat(self.filename).st_mode) # ensure that permissions are at most 640 but respect # original ones if they are tighter backup_perms = orig_perms & ( stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP) # set flag to create file in write only mode flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC try: fd = os.open(backup_file_path, flags, backup_perms) with os.fdopen(fd, 'w') as bf: bf.writelines(lines) except Exception as err: raise GadgetConfigParserError( "Backup file '{0}' is not writable." "".format(backup_file_path), cause=err) f = None if self.output_option_filename: # if output_file was provided create it if not exists prefix, fname = os.path.split(self.output_option_filename) create_option_file(None, fname, prefix, replace=True) # create a copy of the configuration file to output_file shutil.copyfile(self.filename, self.output_option_filename) output_file = self.output_option_filename try: f = open(fs_encode(output_file), 'w') for line in lines: line = fs_decode(line) # empty lines or comment lines are returned unmodified if line.strip() == '' or line.strip().startswith("#") or \ line.strip().startswith(";"): f.write(fs_encode(line)) else: # is it a section header? mo = self._main_opt_parser.SECTCRE.match(line) if mo: if cursect is None: # First section being read cursect = mo.group('header').lower() # Flag section to be dropped if it is not in the # ConfigParser object drop_section = cursect not in in_memory_sections read_sections.add(cursect) else: # we are going into another section, add any # missing options to the previous section unless # it was removed if not drop_section: parser_items = self._main_opt_parser.items( cursect) written_new_options = False for opt, val in parser_items: if opt not in read_options: written_new_options = True f.write(fs_encode(u"{0}\n".format( optval_tostr(opt, val)))) if written_new_options: # add a new line to the end of the new # options f.write("") # get the name of the new section cursect = mo.group('header').lower() # Flag section to be dropped if it is not in the # ConfigParser object drop_section = cursect not in in_memory_sections read_sections.add(cursect) # Reset options read from previous section read_options = set() # write section line if section is not meant to be # removed if not drop_section: f.write(fs_encode(line)) # an option line? else: mo = self._main_opt_parser._optcre.match(line) # if an option inside a section if mo and cursect: if drop_section: # if option belongs to a section to be dropped, # remove it continue optname, _, optval = mo.group('option', 'vi', 'value') x_optname = self._main_opt_parser.optionxform( optname.rstrip()) read_options.add(x_optname) # replace option value if it needs replacing try: new_value = self._main_opt_parser.get( cursect, x_optname) except NoOptionError: # option was removed. Remove it from file continue if new_value == optval: # if value remains the same, leave line as is f.write(fs_encode(line)) continue new_str = optval_tostr(optname, new_value) if not optval: old_str = optname else: # calculate index where option name ends options_ends_at = line.find(optname) + \ len(optname) # calculate index where value ends matched_value_index = line.find( optval, options_ends_at) replace_until_index = matched_value_index + \ len(optval) # we need to replace the old line up until # the value (if exists) old_str = line[:replace_until_index] f.write( fs_encode(line.replace(old_str, new_str))) else: # line format is not valid (neither a section nor # an option nor a comment line) raise GadgetConfigParserError( u"Write operation failed. File '{0}' could " u"not be parsed correctly, parsing error at " u"line '{1}'.".format(output_file, line)) # add missing options for last section (if a section was read) if not drop_section and cursect is not None: parser_items = self._main_opt_parser.items(cursect) written_new_options = False for opt, val in parser_items: if opt not in read_options: # if last line read doesn't have a '\n' at the end, # add it manually. if not line.endswith("\n") and not written_new_options: f.write("\n") written_new_options = True f.write(fs_encode(u"{0}\n".format( optval_tostr(opt, val)))) if written_new_options: # add a new line in case any new options were added to the # last section f.write("") # add new sections and respective options parser_sections = self._main_opt_parser.sections() for s in parser_sections: if s not in read_sections: # if this section was not yet read in the file, it is # a new section. Write it along with its options. f.write(fs_encode(u"[{0}]\n".format(s))) for opt, val in self._main_opt_parser.items(s): f.write(fs_encode(u"{0}\n".format( optval_tostr(opt, val)))) except IOError as err: raise GadgetConfigParserError( u"Option file '{0}' is not writable." u"".format(output_file), cause=err) else: # we've written changes to file, reset modified flag self.modified = False finally: # close the file if f: f.close() mysql_gadgets/common/__init__.py 0000644 00000002241 15231164450 0013010 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # mysql_gadgets/common/constants.py 0000644 00000002644 15231164450 0013274 0 ustar 00 # # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 # """ This module contains constant values to be used throughout the library. """ import os QUOTE_CHAR = "'" if os.name == "posix" else '"' PATH_ENV_VAR = "%PATH%" if os.name == "nt" else "$PATH" HOME_ENV_VAR = "%HOMEPATH%" if os.name == "nt" else "$HOME"
| ver. 1.4 |
Github
|
.
| PHP 8.2.32 | Generation time: 0.1 |
proxy
|
phpinfo
|
Settings