Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-27 02:38
📄 __init__.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from .lve_stats2_lib import get_notification from .ui_config_lib import get_ui_config __all__ = [ "get_notification", "get_ui_config" ]	351 B	2026-02-17 10:40
📁 __pycache__	-	2026-03-27 02:38
📄 cagefs_statistics_config.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # show cagefs usage statistics - work for cloudlinux-config utility import os import subprocess from typing import Optional # NOQA BIN_EXEC = '/usr/sbin/cagefsctl' CAGEFS_STATUS_NOT_INITIALIZED = 'Not initialized' CAGEFS_STATUS_NOT_INSTALLED = 'Not installed' def get_cagefs_user_mode(): # type: () -> Optional[str] """ Get Cagefs user mode, execute cagefsctl --display-user-mode cmd :return string -> cagefs user mode """ if os.path.exists(BIN_EXEC): try: with subprocess.Popen( [BIN_EXEC, '--display-user-mode'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: out, _ = proc.communicate() try: return out.split(':')[1].strip() except IndexError: return None except (OSError, IOError): return None def get_cagefs_status(): """ Get Cagefs user status, execute cagefsctl --cagefs-status cmd :return string -> cagefs status """ if not os.path.exists(BIN_EXEC): return CAGEFS_STATUS_NOT_INSTALLED if not check_cagefs_initialized(): return CAGEFS_STATUS_NOT_INITIALIZED try: with subprocess.Popen( [BIN_EXEC, '--cagefs-status'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: out, _ = proc.communicate() return out.strip() except (OSError, IOError): return None def get_quantity(enabled): """ Get Cagefs quantity if enabled/disabled users, execute cagefsctl --list-enabled cmd for enabled users execute cagefsctl --list-disabled cmd for disabled users :param enabled: true/false flag for quantity enabled/disabled users :return int -> quantity of users """ if os.path.exists(BIN_EXEC): try: if enabled: cmd = '--list-enabled' else: cmd = '--list-disabled' # do not need to check returncode for cagefsctl, not always returns 0 with subprocess.Popen( [BIN_EXEC, cmd], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: out, _ = proc.communicate() try: stat = list(out.split('\n')) if not stat[0]: return 0 return int(stat[0].split(' ')[0]) except (IndexError, ValueError): return None except (OSError, IOError): return None def check_cagefs_initialized(): """ Check if cagefs is initialized with help of --check-cagefs-skeleton command --check-cagefs-skeleton returns rc 0 only if it is initialized """ try: subprocess.check_output([BIN_EXEC, '--check-cagefs-initialized'], stderr=subprocess.STDOUT, shell=False) return True except (subprocess.CalledProcessError, OSError): return False	3.26 KB	2026-02-17 10:40
📄 cl_config_arg_parse.py # -- coding: utf-8 -- # Command line arguments parser for cloudlinux-config utility # cloudlinux-config Utility to check/set different Cloudlinux parameters # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from docopt import DocoptExit, docopt from schema import And, Or, Schema, SchemaError from clcommon.cpapi import is_reseller def parse_cloudlinux_config_opts(argv, _is_json_need=False): # program name prog_name = "cloudlinux-config" docstring = f"""Utility to get/set any Cloudlinux config Usage: {prog_name} [get] [--json] [--for-reseller <str>] {prog_name} [set] [--json] --data <str> [--for-reseller <str>] {prog_name} [set] [--json] --reset-inodes-limit {prog_name} (-h \| --help) Options: --json Return data in JSON format. --data <str> Set parameters from JSON str --reset-inodes-limit Reset inodes limits --for-reseller <str> Set/Get configuration options for specific reseller (only lve-stats) -h, --help Show this help message and exit """ try: args = docopt(docstring, argv) except DocoptExit: s_error_string = 'ERROR: Invalid parameter passed' if not _is_json_need: s_error_string += "\n\n" + docstring return False, s_error_string if not args["get"] and not args["set"]: args["get"] = True # uncomment this for add full edit mode # gov_int_list = And(Use(lambda x:x.split(",")), lambda x:len(x) == 4, # Use(lambda x:map(int, x))) s = Schema({ "set": bool, "get": bool, "--json": And(bool, lambda x: x, error="use --json option, other modes currently unsupported"), "--data": Or(None, str), "--for-reseller": And(Or(None, str), lambda x: x is None or is_reseller(x), error="non-existent reseller name"), "--reset-inodes-limit": bool, "--help": bool, }) try: args = s.validate(args) status = True except SchemaError as e: args = str(e) status = False return status, args	2.23 KB	2026-02-17 10:40
📄 clconfig_utils.py # -- coding: utf-8 -- # cl_config_utils - some helper functions for cloudlinux-config utility # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT def repack_dict(keys_dict, source_dict, func_convertor=None, default=False): result_dict = {} if func_convertor is None: def func_convertor(x): return int(x) for key, value in keys_dict.items(): if key in source_dict: result_dict[value] = func_convertor(source_dict[key]) else: result_dict[value] = default return result_dict def min_num_notify_converter(val): try: result = int(val) except (ValueError, TypeError): return 1 return result if result > 0 else 1 def time_convertor_to_dict(time_string): try: period = int(time_string) units = 'minutes' except (ValueError, TypeError): try: period = int(time_string[:-1]) if 'h' in time_string: units = 'hours' elif 's' in time_string: units = 'seconds' elif 'd' in time_string: units = 'days' else: units = 'minutes' except TypeError: return None result_dict = {'period': period, 'unitOfTime': units} return result_dict def str_to_boolean(string): if string.lower() in ['y', '1', 'on', 'true', 'yes']: return True return False def str_to_int(string): """ Converts (if possible) string value to int :return: None - if string value isn't convertable, int - success """ try: return int(string) except (ValueError, TypeError): return None def boolean_to_yes_no(bool_val): if not isinstance(bool_val, bool): return None if bool_val: return 'Y' return 'N' def boolean_to_on_off(bool_val): if not isinstance(bool_val, bool): return None if bool_val: return 'on' return 'off' def boolean_to_0_1(bool_val): if not isinstance(bool_val, bool): return None if bool_val: return '1' return '0' def is_int(int_val): if not isinstance(int_val, int): return None if int_val < 0: return None return int_val def is_positive_int(int_val): if not isinstance(int_val, int): return None if int_val < 1: return None return int_val def time_unit_to_letter(unit): ret_dict = {'days': 'd', 'hours': 'h', 'minutes': 'm', 'seconds': 's'} if unit not in ret_dict: return None return ret_dict[unit] def is_string(str_val): return isinstance(str_val, str) or None def is_string_or_none(str_val): if is_string(str_val) or str_val is None: return True return None	2.84 KB	2026-02-17 10:40
📄 config.py	23.64 KB	2026-02-17 10:40
📄 db_governor_lib.py # -- coding: utf-8 -- # db_governer_lib.py - Library for MySQL governor config files manipulating for cloudlinux-config utility # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import os import subprocess from typing import Optional # NOQA from lxml import etree from clcommon.clexception import FormattedException from clcommon.lib.mysql_governor_lib import MySQLGovernor from .clconfig_utils import ( boolean_to_on_off, str_to_boolean, time_convertor_to_dict, time_unit_to_letter, ) class DBGovernorException(FormattedException): pass DEFAULTS_DICT = { 'modeOfOperation': 'off', 'gatherDataForDetailedStats': 'on', 'save_uid': 'off', 'logRestrictedUsersQueries': 'off', 'scriptPath': '', 'userMaxConnections': 30, 'restrictType': { 'mode': 'limit', 'unlimit': '60s' }, 'slowQueries': { 'kill': 'off', 'logPath': '', 'timeout': 30 }, 'restrictLog': { 'logPath': '', 'format': 'LONG' }, 'errorLog': { 'logPath': '', 'format': 'ERROR' } } DbGovConfig = '/etc/container/mysql-governor.xml' # For unit tests def _open(file_name): return open(file_name, encoding='utf-8') def get_data_from_xml(x, node, attrib, default=None): """ Gets data from specified node and attribute :param x: Parent node object :param node: Node name to extract data :param attrib: Attribute name to extract data :param default: Default value to return if specified node/attribute not found :return: Value """ try: path = '//' + str(node) + '/@' + str(attrib) data = x.xpath(path)[0] except (AttributeError, IndexError): return default return data def convert_time_value_to_seconds(value): """ Convert value in minutes, hours, days to seconds :param value: value of slow limit :return: value in seconds """ value = str(value).lower() if value.endswith('m'): result = int(value[:-1]) * 60 elif value.endswith('h'): result = int(value[:-1]) * 60 * 60 elif value.endswith('d'): result = int(value[:-1]) * 60 * 60 * 24 elif value.endswith('s'): result = int(value[:-1]) else: result = int(value) return result def get_slow_queries_timeout(x, default): """ Gets slow queries timeout :param x: Root node :param default: Default value to return if timeout not found :return: Timeout """ try: data = x.xpath("default/limit[@name='slow']/@current")[0] data = convert_time_value_to_seconds(data) return int(data) except (AttributeError, IndexError, ValueError): return default def set_slow_queries_timeout(node, slow_timeout): """ Sets slow queries timeout :param node: Root node :param slow_timeout: new timeout value :return: None """ path = "default/limit[@name='slow']" tag_list = node.xpath(path) if len(tag_list) == 0: # Create default/limit chain node_list = node.xpath('default') if len(node_list) == 0: new_node = etree.Element('default') node.append(new_node) # 'default' node present, but 'limit' node with name="slow" attribute is absent default_node = node.xpath('default')[0] # Create 'limit' node tag_to_set = etree.Element('limit') default_node.append(tag_to_set) # set name attribute tag_to_set.set('name', 'slow') else: tag_to_set = tag_list[0] # Set attribute value tag_to_set.set('current', str(slow_timeout)) def set_data_to_xml(x, node_name, attribute_name, value): """ Sets value to specified node/attribute :param x: Parent node :param node_name: Node name :param attribute_name: Attribute name :param value: Value to set :return: None """ path = '//' + str(node_name) tag_list = x.xpath(path) if len(tag_list) == 0: # Tag absent, add it new_node = etree.Element(node_name) x.append(new_node) tag_list = x.xpath(path) tag = tag_list[0] # Set specified attribute tag.set(attribute_name, value) def _get_timedict_from_xml(x, tag_name, attr_name): """ Get time from specified tag/attribute as seconds :param x: XML object :param tag_name: XML tag name :param attr_name: XML object name :return: Number of seconds. Throws DBGovernorException if any error """ value = get_data_from_xml(x, tag_name, attr_name) try: return time_convertor_to_dict(value) except Exception as e: raise DBGovernorException( f"{DbGovConfig} has invalid value in governer/{tag_name}/@{attr_name}" ) from e def get_gov_mode_operation(parsed_xml_tree=None): # type: (etree.ElementBase) -> Optional[str] """ Get MySQL Governor mode. We return None if governor config isn't exists All exceptions will be catched in top level functions :param parsed_xml_tree: parsed governor config """ if not os.path.exists(DbGovConfig): return None if parsed_xml_tree is None: f = _open(DbGovConfig) x = etree.parse(f).getroot() else: x = parsed_xml_tree return get_data_from_xml(x, 'lve', 'use', DEFAULTS_DICT['modeOfOperation']) def get_db_gov_config(): """ Get MySQL Governer config :return: Dictionary with config data """ try: if not os.path.exists(DbGovConfig): return {} f = _open(DbGovConfig) x = etree.parse(f).getroot() result_dict = {} result_dict['modeOfOperation'] = get_gov_mode_operation(parsed_xml_tree=x) result_dict['gatherDataForDetailedStats'] = str_to_boolean( get_data_from_xml(x, 'statistic', 'mode', DEFAULTS_DICT['gatherDataForDetailedStats'])) result_dict['logRestrictedUsersQueries'] = str_to_boolean( get_data_from_xml(x, 'logqueries', 'use', DEFAULTS_DICT['logRestrictedUsersQueries'])) result_dict['scriptPath'] = get_data_from_xml(x, 'restrict', 'script', DEFAULTS_DICT['scriptPath']) result_dict['userMaxConnections'] = int( get_data_from_xml(x, 'restrict', 'user_max_connections', DEFAULTS_DICT['userMaxConnections'])) result_dict['restrictedTimePeriods'] = { 'level1': _get_timedict_from_xml(x, 'restrict', 'level1'), 'level2': _get_timedict_from_xml(x, 'restrict', 'level2'), 'level3': _get_timedict_from_xml(x, 'restrict', 'level3'), 'level4': _get_timedict_from_xml(x, 'restrict', 'level4'), 'timeout': _get_timedict_from_xml(x, 'restrict', 'timeout') } result_dict['restrictType'] = { 'mode': get_data_from_xml(x, 'restrict_mode', 'use', DEFAULTS_DICT['restrictType']['mode']), 'unlimit': time_convertor_to_dict( get_data_from_xml(x, 'restrict_mode', 'unlimit', DEFAULTS_DICT['restrictType']['unlimit'])) } result_dict['slowQueries'] = {'kill': str_to_boolean(get_data_from_xml(x, 'slow_queries', 'run', DEFAULTS_DICT['slowQueries']['kill'])), 'logPath': get_data_from_xml(x, 'slow_queries', 'log', DEFAULTS_DICT['slowQueries']['logPath']), 'timeout': get_slow_queries_timeout(x, DEFAULTS_DICT['slowQueries']['timeout']) } result_dict['restrictLog'] = { 'logPath': get_data_from_xml(x, 'restrict', 'log', DEFAULTS_DICT['restrictLog']['logPath']), 'format': get_data_from_xml(x, 'restrict', 'format', DEFAULTS_DICT['restrictLog']['format']) } result_dict['errorLog'] = {'logPath': get_data_from_xml(x, 'log', 'file', DEFAULTS_DICT['errorLog']['logPath']), 'level': get_data_from_xml(x, 'log', 'mode', DEFAULTS_DICT['errorLog']['format']) } result_dict.update(get_db_gov_stat_info()) return {'mySQLGovSettings': result_dict} except (OSError, IOError, etree.XMLSyntaxError, etree.XMLSchemaParseError, etree.XMLSchemaError, DBGovernorException) as e: raise DBGovernorException(f"{DbGovConfig} read/parse error: {e}") from e def set_db_gov_config(parameters_dict): """ Set MySQL governer config :param parameters_dict: Parameters to set dictionary. For example: { u'modeOfOperation': u'abusers', u'scriptPath': u'', u'gatherDataForDetailedStats': True, u'userMaxConnections': 30, u'logRestrictedUsersQueries': False, u'restrictLog': {u'logPath': u'/var/log/dbgovernor-restrict.log', u'format': u'LONG'}, u'restrictedTimePeriods': {u'level1': {u'unitOfTime': u'seconds', u'period': 60}, u'level2': {u'unitOfTime': u'minutes', u'period': 15}, u'level3': {u'unitOfTime': u'hours', u'period': 1}, u'level4': {u'unitOfTime': u'days', u'period': 1}, u'timeout': {u'unitOfTime': u'hours', u'period': 1}}, u'errorLog': {u'logPath': u'/var/log/dbgovernor-error-my.log', u'level': u'DEBUG'}, u'slowQueries': {u'logPath': u'/var/log/dbgovernor-kill.log', u'kill': True, u'timeout': 30}, u'restrictType': {u'mode': u'limit', u'unlimit': {u'unitOfTime': u'seconds', u'period': 60}} } Any key in dictionary can absent, appropriate MySQL Governor parameter will not changed :return: None. Throws DBGovernorException if error """ try: # Do nothing if MySQL Governor config does not exist if not os.path.exists(DbGovConfig): return # Open config and parse it f = _open(DbGovConfig) x = etree.parse(f).getroot() f.close() ################################## # 1. Process single node/attribute # Control data list struct: # (input_dictionary_keys, xml_tag_name, xml_attrubute_name) control_data_list = [ ("['modeOfOperation']", 'lve', 'use', str), # lve/@use ("['scriptPath']", 'restrict', 'script', str), # restrict/@script ("['gatherDataForDetailedStats']", 'statistic', 'mode', boolean_to_on_off), # statistic/@mode ("['userMaxConnections']", 'restrict', 'user_max_connections', str), # restrict/@user_max_connections ("['logRestrictedUsersQueries']", 'logqueries', 'use', boolean_to_on_off), # logqueries/@use ("['restrictLog']['logPath']", 'restrict', 'log', str), # restrict/@log ("['restrictLog']['format']", 'restrict', 'format', str), # restrict/@format ("['errorLog']['logPath']", 'log', 'file', str), # log/@file ("['errorLog']['level']", 'log', 'mode', str), # log/@mode ("['slowQueries']['logPath']", 'slow_queries', 'log', str), # slow_queries/@log ("['slowQueries']['kill']", 'slow_queries', 'run', boolean_to_on_off), # slow_queries/@run ("['restrictType']['mode']", 'restrict_mode', 'use', str) # restrict_mode/@use ] for control_data_item in control_data_list: try: input_dict_keys, xml_tag_name, xml_attribute_name, func_converter = control_data_item # Get input dictionary value. Generates KeyError exception if key(s) not exist val = eval(f"parameters_dict{input_dict_keys}") set_data_to_xml(x, xml_tag_name, xml_attribute_name, func_converter(val)) except KeyError: pass ################################## # 2. Process values pairs # Item structure: (period_indexes, period_unit_of_time, xml_tag_name, xml_attrubute_name) control_data_pairs_list = [ ("['restrictType']['unlimit']['period']", "['restrictType']['unlimit']['unitOfTime']", 'restrict_mode', 'unlimit'), # restrict_mode/@unlimit ("['restrictedTimePeriods']['level1']['period']", "['restrictedTimePeriods']['level1']['unitOfTime']", 'restrict', 'level1'), # restrict/@level1 ("['restrictedTimePeriods']['level2']['period']", "['restrictedTimePeriods']['level2']['unitOfTime']", 'restrict', 'level2'), # restrict_mode/@level2 ("['restrictedTimePeriods']['level3']['period']", "['restrictedTimePeriods']['level3']['unitOfTime']", 'restrict', 'level3'), # restrict_mode/@level3 ("['restrictedTimePeriods']['level4']['period']", "['restrictedTimePeriods']['level4']['unitOfTime']", 'restrict', 'level4'), # restrict_mode/@level4 ("['restrictedTimePeriods']['timeout']['period']", "['restrictedTimePeriods']['timeout']['unitOfTime']", 'restrict', 'timeout'), # restrict_mode/@timeout ] for control_data_item in control_data_pairs_list: period_indexes, period_unit_of_time_indexes, xml_tag_name, xml_attribute_name = control_data_item try: # period_indexes, period_unit_of_time_indexes, xml_tag_name, xml_attribute_name = control_data_item # Get input dictionary value. Generates KeyError exception if key(s) not exist period_val = eval(f"parameters_dict{period_indexes}") period_unit_of_time = time_unit_to_letter(eval(f"parameters_dict{period_unit_of_time_indexes}")) set_data_to_xml(x, xml_tag_name, xml_attribute_name, f"{period_val}{period_unit_of_time}") except KeyError: pass ################################## # 3. Process special values # Write slow queries timeout try: slow_queries_timeout = parameters_dict['slowQueries']['timeout'] set_slow_queries_timeout(x, slow_queries_timeout) except KeyError: pass ################################## # 4. Save result s_xml = etree.tostring(x) with open(DbGovConfig, 'wb') as f: f.write(b'<?xml version="1.0" ?>\n%s\n' % s_xml) except (OSError, IOError, etree.XMLSyntaxError, etree.XMLSchemaParseError, etree.XMLSchemaError) as e: raise DBGovernorException({'message': "%(dbgov_cfg)s error: " + str(e), 'context': {'dbgov_cfg': DbGovConfig}}) from e ################################## # 5. Restart governer # /sbin/service db_governor restart subprocess.run('/sbin/service db_governor restart &>/dev/null &', shell=True, executable='/bin/bash', check=False) def get_db_gov_stat_info(): """ Get db_governor status and version info :return: dict( 'dbGovVersion': <version>, 'dbGovStatus': enabled\|error\|disabled, 'dbGovError': error details ) """ db_gov = MySQLGovernor() status, error = db_gov.get_governor_status() return { 'dbGovVersion': db_gov.get_governor_version(), 'dbGovStatus': status, 'dbGovError': str(error) }	15.19 KB	2026-02-17 10:40
📄 db_info_lib.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import os import re from clcommon.utils import ExternalProgramFailed, run_command MYSQL_BINARIES = [ "/usr/sbin/mysqld", "/usr/libexec/mysqld", # Fallback option "/usr/local/bin/mysqld", # Present on CL9+DA ] class MysqlInfo: def __init__(self): self.mysqld_v = self.retrieve_server_info() @staticmethod def get_binary(): """ Find mysql server binary :return: path to mysql binary """ try: for binary in MYSQL_BINARIES: if os.path.exists(binary): return binary return MYSQL_BINARIES[1] except OSError: return MYSQL_BINARIES[1] def retrieve_server_info(self): """ Get server information through `<mysql_server_binary> -V` command :return: full command output """ try: mysql_binary = self.get_binary() return run_command([mysql_binary, "-V"]) except ExternalProgramFailed: # may be no such file or unknown error happen return None def get(self): """ Return gathered data :return: dict( 'vendor': MySQL\|MariaDB\|Percona, 'version': server version in the form of {major}.{minor} or {major}.{minor}-{release} for percona, 'cll-lve': patches from CL applied or not (True\|False) ) """ return {"vendor": self.get_vendor(), "version": self.get_version(), "cll-lve": self.is_patched()} def get_vendor(self): """ Extract MySQL vendor from server info :return: MySQL\|MariaDB\|Percona or `unknown` if failed to apply regex or None if there is no server info (this usually means that there are no MySQL installed) """ if self.mysqld_v is None: return None # regex to find vendor -- something, starting from round bracket and followed by `Server` # (see test_clconfig_db_info_lib.py for detailed examples) p = re.compile(r"(?<=\().+(?=\sServer)") try: # we need only first word of first element return p.findall(self.mysqld_v)[0].split()[0] except IndexError: return "unknown" def get_version(self): """ Retrieve MySQL server version from server info :return: X.X.X e.g. 10.2.16\|5.6.39, includes release for percona (X.X.X-X.X e.g. 5.6.40-84.0) or `unknown` if failed to apply regex or None if there is no server info (this usually means that there are no MySQL installed) """ if self.mysqld_v is None: return None # regex to find version -- numbers, divided by `.` or `-` which go after `Ver` # (see test_clconfig_db_info_lib.py for detailed examples) p = re.compile(r"(?<=Ver\s)[0-9\.\-]+") try: # we need first element, also should sanitize from trailing `-` return p.findall(self.mysqld_v)[0].rstrip("-") except IndexError: return "unknown" def is_patched(self): """ Retrieve information about CL patch by `cll-lve` in server info :return: True if contains `cll-lve` seq False otherwise or None if there is no server info (this usually means that there are no MySQL installed) """ if self.mysqld_v is None: return None return "cll-lve" in self.mysqld_v	3.63 KB	2026-02-17 10:40
📄 lve_stats2_lib.py # -- coding: utf-8 -- # lve_stats2_lib.py - lve-stats library for cloudlinux-config library # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import os import subprocess from clcommon.clexception import FormattedException from clcommon.utils import mod_makedirs from .clconfig_utils import ( boolean_to_yes_no, is_positive_int, min_num_notify_converter, repack_dict, str_to_boolean, time_convertor_to_dict, time_unit_to_letter, ) class LveStats2Exception(FormattedException): pass StatsNotifierDir = '/etc/sysconfig/lvestats.config' StatsNotifierConfig = StatsNotifierDir + '/StatsNotifier.cfg' notify_dict = {'NOTIFY_ADMIN': 'notifyAdmin', 'NOTIFY_RESELLER': 'notifyResellers', 'NOTIFY_CUSTOMER': 'notifyCustomers', 'NOTIFY_INCLUDE_RESELLER_CUSTOMER': 'notifyResellerCustomers'} faults_dict = {'NOTIFY_MEMORY': 'mem', 'NOTIFY_IOPS': 'iops', 'NOTIFY_NPROC': 'nproc', 'NOTIFY_IO': 'io', 'NOTIFY_EP': 'concurrentConnections', 'NOTIFY_CPU': 'cpu'} email_notify_dict = {'NOTIFY_FROM_EMAIL': 'notifyFromEmail', 'REPORT_ADMIN_EMAIL': 'reportAdminMail', 'NOTIFY_FROM_SUBJECT': 'notifySubject', 'NOTIFY_CHARSET_EMAIL': 'notifyCharset'} number_notify_dict = {'NOTIFY_MIN_FAULTS_USER': 'user', 'NOTIFY_MIN_FAULTS_ADMIN': 'admin'} notify_time_dict = {'NOTIFY_INTERVAL_ADMIN': 'admin', 'NOTIFY_INTERVAL_USER': 'user'} # For unit tests def _open(file_name, mode='r'): return open(file_name, mode=mode, encoding='utf-8') def get_notification(reseller=None): """ Retrieves lve-stats2 notifications parameters :param str reseller: reseller name, if we want notifications parameters for a reseller :return: dict. For example: {'faultsNotification': {'notifyResellerCustomers': True, 'notifyResellers': True, 'minimumNumberOfFaultsToNotify': {'admin': 0, 'user': 0}, 'notifyAdmin': True, 'notify': {'admin': {'unitOfTime': 'hours', 'period': 12}, 'user': {'unitOfTime': 'hours', 'period': 12} }, 'faultsToInclude': {'mem': True, 'iops': False, 'io': False, 'nproc': False, 'concurrentConnections': False, 'cpu': True}, 'notifyCustomers': False } } """ config_path = StatsNotifierConfig if not os.path.exists(config_path): return {} try: f = _open(config_path) config = f.readlines() f.close() except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s read error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e config_dict = {} for line in config: # skip comments and empty lines if line.startswith('#') or line.strip() == '': continue key, value = line.split('=') config_dict[key] = value.strip() notify_d = notify_dict faults_d = faults_dict number_notify_d = number_notify_dict notify_time_d = notify_time_dict faults_notification_dict = repack_dict(notify_d, config_dict, str_to_boolean) faults_include_dict = repack_dict(faults_d, config_dict, str_to_boolean) min_num_notify = repack_dict(number_notify_d, config_dict, min_num_notify_converter, default=1) period_notify = repack_dict(notify_time_d, config_dict, time_convertor_to_dict) email_settings = repack_dict(email_notify_dict, config_dict, str, default=None) faults_notification_dict['faultsToInclude'] = faults_include_dict faults_notification_dict['minimumNumberOfFaultsToNotify'] = min_num_notify faults_notification_dict['notify'] = period_notify faults_notification_dict['email'] = email_settings return {"faultsNotification": faults_notification_dict} def set_notification(parameters_dict, reseller=None): """ Sets lve-stats2 notifications parameters :param parameters_dict: Parametres to set. For example: {u'notifyResellers': True, u'faultsToInclude': {u'mem': True, u'iops': False, u'io': False, u'nproc': False, u'concurrentConnections': False, u'cpu': True}, u'minimumNumberOfFaultsToNotify': {u'admin': 0, u'user': 0}, u'notifyAdmin': True, u'notify': {u'admin': {u'unitOfTime': u'hours', u'period': 12}, u'user': {u'unitOfTime': u'hours', u'period': 12} }, u'notifyResellerCustomers': True, u'notifyCustomers': False } :param str reseller: either reseller name or None """ # Do nothing, if admin's lvestats2 notifier config not found # either parameters_dict is empty and we shouldn't change anything if not os.path.exists(StatsNotifierConfig) or len(parameters_dict) == 0: return notifier_params = {} config_path = StatsNotifierConfig notifier_params = get_notification() # create directory and empty config if not exists if not os.path.exists(os.path.dirname(config_path)): mod_makedirs(os.path.dirname(config_path), 0o755) if not os.path.exists(config_path): _open(config_path, 'w').close() # Read notifier config try: f = _open(config_path) config_lines = [line.strip() for line in f.readlines()] f.close() except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s read error: " + str(e), 'context': {'lvestats_cfg': StatsNotifierConfig}}) from e # Fill parameters to change lve-stats config faults_config_map = {'NOTIFY_CPU': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/cpu'), 'NOTIFY_NPROC': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/nproc'), 'NOTIFY_IO': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/io'), 'NOTIFY_MEMORY': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/mem'), 'NOTIFY_EP': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/concurrentConnections'), 'NOTIFY_IOPS': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/iops')} faults_config_map.update({ 'NOTIFY_INTERVAL_ADMIN': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/admin/period', 'notify/admin/unitOfTime'), 'NOTIFY_INTERVAL_USER': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/user/period', 'notify/user/unitOfTime'), 'NOTIFY_ADMIN': (lambda x: boolean_to_yes_no(x), 'notifyAdmin'), 'NOTIFY_CUSTOMER': (lambda x: boolean_to_yes_no(x), 'notifyCustomers'), 'NOTIFY_RESELLER': (lambda x: boolean_to_yes_no(x), 'notifyResellers'), 'NOTIFY_INCLUDE_RESELLER_CUSTOMER': (lambda x: boolean_to_yes_no(x), 'notifyResellerCustomers'), 'NOTIFY_MIN_FAULTS_ADMIN': 'minimumNumberOfFaultsToNotify/admin', 'NOTIFY_MIN_FAULTS_USER': 'minimumNumberOfFaultsToNotify/user'}) # Update dict with email parameters # If None is passed, line will be removed from config before writing it back to file faults_config_map.update({ 'NOTIFY_FROM_EMAIL': 'email/notifyFromEmail', 'NOTIFY_FROM_SUBJECT': 'email/notifySubject', 'REPORT_ADMIN_EMAIL': 'email/reportAdminMail', 'NOTIFY_CHARSET_EMAIL': 'email/notifyCharset'}) new_config = {} def get_val_by_path(path): path_parts = path.split('/') try: point = parameters_dict for part in path_parts: point = point[part] except KeyError: point = notifier_params['faultsNotification'] for part in path_parts: point = point[part] return point for key, val in faults_config_map.items(): try: if isinstance(val, str): new_config[key] = get_val_by_path(val) else: new_config[key] = val[0](*(get_val_by_path(x) for x in val[1:])) except KeyError: continue # to do not change iter object while iterate it config_to_write = [line.strip() for line in config_lines] try: for idx, line in enumerate(config_lines): # Pass comment and empty lines if line.startswith('#') or line.strip() == '': continue try: key, value = line.split('=') except ValueError as e: raise LveStats2Exception({ 'message': '%(lvestats_cfg)s format error: ' + f'{idx + 1}: {line}', 'context': {'lvestats_cfg': config_path}}) from e if key in new_config: # Parameter must be changed config_to_write[idx] = f"{key}={new_config[key]}" # Remove used parameter del new_config[key] except (KeyError, IndexError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s format error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e # Add absent options for key, value in new_config.items(): config_to_write.append(f"{key}={value}") # Email notifications are optional # They should be removed from config, if None value is passed def filter_config_lines(_line): if _line.startswith('#') or _line.strip() == '': return True _, _value = _line.split('=') return _value != 'None' config_to_write = list(filter(filter_config_lines, config_to_write)) # Write config back to file try: with open(config_path, 'w', encoding='utf-8') as f: f.writelines('\n'.join(config_to_write)) except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s write error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e # Restart lvestats server # /sbin/service lvestats reload subprocess.run('/sbin/service lvestats reload &>/dev/null &', shell=True, executable='/bin/bash', check=False)	10.63 KB	2026-02-17 10:40
📄 lve_stats2_reseller_lib.py # -- coding: utf-8 -- # lve_stats2_reseller_lib.py - lve-stats reseller library for cloudlinux-config library # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import os import subprocess from clcommon.utils import mod_makedirs from . import lve_stats2_lib from .clconfig_utils import ( boolean_to_yes_no, is_positive_int, min_num_notify_converter, repack_dict, str_to_boolean, time_convertor_to_dict, time_unit_to_letter, ) from .lve_stats2_lib import LveStats2Exception, StatsNotifierConfig, StatsNotifierDir, faults_dict ResellerStatsNotifierDir = StatsNotifierDir + '/resellers.d/' reseller_notify_dict = { 'NOTIFY_RESELLER_ON_TOTAL_FAULTS': 'notifyReseller', 'NOTIFY_RESELLER_ON_CUSTOMERS_FAULTS': 'notifyResellerOnCustomers', 'NOTIFY_CUSTOMERS_ON_FAULTS': 'notifyCustomers' } reseller_number_notify_dict = { 'NOTIFY_MIN_FAULTS_RESELLER': 'reseller', 'NOTIFY_MIN_FAULTS_CUSTOMER': 'customer' } reseller_notify_time_dict = { 'NOTIFY_INTERVAL_RESELLER': 'reseller', 'NOTIFY_INTERVAL_CUSTOMER': 'customer' } # For unit tests def _open(file_name, mode='r'): return open(file_name, mode=mode, encoding='utf-8') def _create_default_reseller_config(): """ Creates default StatsNotifier config for reseller """ admin_config = lve_stats2_lib.get_notification().get('faultsNotification') admin_faults = {key: admin_config['faultsToInclude'][key] for key in ( 'mem', 'iops', 'io', 'nproc', 'concurrentConnections', 'cpu' )} return {'faultsNotification': {'notifyResellerOnCustomers': admin_config['notifyResellers'], 'minimumNumberOfFaultsToNotify': { 'reseller': admin_config['minimumNumberOfFaultsToNotify']['admin'], 'customer': admin_config['minimumNumberOfFaultsToNotify']['user']}, 'notifyReseller': admin_config['notifyResellers'], 'notify': {'reseller': {'unitOfTime': admin_config['notify']['admin']['unitOfTime'], 'period': admin_config['notify']['admin']['period']}, 'customer': {'unitOfTime': admin_config['notify']['user']['unitOfTime'], 'period': admin_config['notify']['user']['period']} }, 'faultsToInclude': admin_faults, 'notifyCustomers': admin_config['notifyResellerCustomers'] }} def get_notification(reseller, defaults=False): """ Retrieves lve-stats2 notifications parameters for resellers :param bool defaults: True if we want get defaults when config is not set, else False :param str reseller: reseller name, if we want notifications parameters for a reseller :return: dict. For example: {'faultsNotification': {'notifyResellerCustomers': True, 'notifyResellers': True, 'minimumNumberOfFaultsToNotify': {'admin': 0, 'user': 0}, 'notifyAdmin': True, 'notify': {'admin': {'unitOfTime': 'hours', 'period': 12}, 'user': {'unitOfTime': 'hours', 'period': 12} }, 'faultsToInclude': {'mem': True, 'iops': False, 'io': False, 'nproc': False, 'concurrentConnections': False, 'cpu': True}, 'notifyCustomers': False } } """ if reseller is None: raise ValueError('reseller can`t be None') config_path = os.path.join(ResellerStatsNotifierDir, f"{reseller}_StatsNotifier.cfg") if not os.path.exists(config_path): return _create_default_reseller_config() if defaults else {} try: f = _open(config_path) config = f.readlines() f.close() except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s read error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e config_dict = {} for line in config: if line.startswith('#') or line.strip() == '': continue key, value = line.split('=') config_dict[key] = value.strip() notify_d = reseller_notify_dict faults_d = faults_dict number_notify_d = reseller_number_notify_dict notify_time_d = reseller_notify_time_dict faults_notification_dict = repack_dict(notify_d, config_dict, str_to_boolean) faults_include_dict = repack_dict(faults_d, config_dict, str_to_boolean) min_num_notify = repack_dict(number_notify_d, config_dict, min_num_notify_converter, default=1) period_notify = repack_dict(notify_time_d, config_dict, time_convertor_to_dict) faults_notification_dict['faultsToInclude'] = faults_include_dict faults_notification_dict['minimumNumberOfFaultsToNotify'] = min_num_notify faults_notification_dict['notify'] = period_notify return {"faultsNotification": faults_notification_dict} def set_notification(parameters_dict, reseller): """ Sets lve-stats2 notifications parameters :param parameters_dict: Parametres to set. For example: {u'notifyResellers': True, u'faultsToInclude': {u'mem': True, u'iops': False, u'io': False, u'nproc': False, u'concurrentConnections': False, u'cpu': True}, u'minimumNumberOfFaultsToNotify': {u'admin': 0, u'user': 0}, u'notifyAdmin': True, u'notify': {u'admin': {u'unitOfTime': u'hours', u'period': 12}, u'user': {u'unitOfTime': u'hours', u'period': 12} }, u'notifyResellerCustomers': True, u'notifyCustomers': False } :param str reseller: either reseller name or None """ # Do nothing, if admin's lvestats2 notifier config not found # either parameters_dict is empty and we shouldn't change anything if not os.path.exists(StatsNotifierConfig) or len(parameters_dict) == 0: return notifier_params = {} if reseller is None: raise ValueError('reseller can`t be None') config_path = os.path.join(ResellerStatsNotifierDir, f"{reseller}_StatsNotifier.cfg") if not os.path.exists(config_path): notifier_params.update(_create_default_reseller_config()) else: notifier_params.update(get_notification(reseller=reseller)) # create directory and empty config if not exists if not os.path.exists(os.path.dirname(config_path)): mod_makedirs(os.path.dirname(config_path), 0o755) if not os.path.exists(config_path): _open(config_path, 'w').close() # Read notifier config try: f = _open(config_path) config_lines = [line.strip() for line in f.readlines()] f.close() except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s read error: " + str(e), 'context': {'lvestats_cfg': StatsNotifierConfig}}) from e # Fill parameters to change lve-stats config faults_config_map = {'NOTIFY_CPU': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/cpu'), 'NOTIFY_NPROC': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/nproc'), 'NOTIFY_IO': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/io'), 'NOTIFY_MEMORY': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/mem'), 'NOTIFY_EP': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/concurrentConnections'), 'NOTIFY_IOPS': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/iops')} faults_config_map.update({ 'NOTIFY_INTERVAL_RESELLER': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/reseller/period', 'notify/reseller/unitOfTime'), 'NOTIFY_INTERVAL_CUSTOMER': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/customer/period', 'notify/customer/unitOfTime'), 'NOTIFY_RESELLER_ON_TOTAL_FAULTS': (lambda x: boolean_to_yes_no(x), 'notifyReseller'), 'NOTIFY_RESELLER_ON_CUSTOMERS_FAULTS': (lambda x: boolean_to_yes_no(x), 'notifyResellerOnCustomers'), 'NOTIFY_CUSTOMERS_ON_FAULTS': (lambda x: boolean_to_yes_no(x), 'notifyCustomers'), 'NOTIFY_MIN_FAULTS_RESELLER': 'minimumNumberOfFaultsToNotify/reseller', 'NOTIFY_MIN_FAULTS_CUSTOMER': 'minimumNumberOfFaultsToNotify/customer'}) new_config = {} def get_val_by_path(path): path_parts = path.split('/') try: point = parameters_dict for part in path_parts: point = point[part] except KeyError: point = notifier_params['faultsNotification'] for part in path_parts: point = point[part] return point for key, val in faults_config_map.items(): try: if isinstance(val, str): new_config[key] = get_val_by_path(val) else: new_config[key] = val[0](*(get_val_by_path(x) for x in val[1:])) except KeyError: continue # to do not change iter object while iterate it config_to_write = [line.strip() for line in config_lines] try: for idx, line in enumerate(config_lines): # Pass comment lines if line.startswith('#') or line.strip() == '': continue try: key, value = line.split('=') except ValueError as e: raise LveStats2Exception({ 'message': '%(lvestats_cfg)s format error: ' + f'{idx + 1}: {line}', 'context': {'lvestats_cfg': config_path}}) from e if key in new_config: # Parameter must be changed config_to_write[idx] = f"{key}={new_config[key]}" # Remove used parameter del new_config[key] except (KeyError, IndexError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s format error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e # Add absent options for key, value in new_config.items(): config_to_write.append(f"{key}={value}") # Write config back to file try: with open(config_path, 'w', encoding='utf-8') as f: f.writelines('\n'.join(config_to_write)) except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s write error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e # Restart lvestats server # /sbin/service lvestats reload subprocess.run('/sbin/service lvestats reload &>/dev/null &', shell=True, executable='/bin/bash', check=False)	11.05 KB	2026-02-17 10:40
📄 ui_config_lib.py # -- coding: utf-8 -- # ui_config_lib.py - Library for panel config files manipulating for cloudlinux-config utility # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import subprocess import cldetectlib as detect from clcommon.clexception import FormattedException from clcommon.clquota import check_quota_enabled from clcommon.const import Feature from clcommon.cpapi import is_panel_feature_supported from clcommon.ui_config import UIConfig from cllimits.lib import exec_utility from .clconfig_utils import boolean_to_0_1 class UIConfigException(FormattedException): pass DYNAMIC_UI_CTL_CMD = '/usr/share/l.v.e-manager/utils/dynamicui.py' def get_ui_config(): """ Retrives UI settings from config file :return: dict. For example: {'inodeLimits': {'showUserInodesUsage': False}, 'uiSettings': {'hideRubyApp': False, 'hideLVEUserStat': False, 'hidePythonApp': False, 'hidePHPextensions': True} } """ return filter_flags(UIConfig().get_config()) def refresh_panel(): try: retcode, out, err = exec_utility(DYNAMIC_UI_CTL_CMD, ['--sync-conf=all'], stderr=True) except OSError as e: raise UIConfigException({'message': "Can't execute file %(ctl_cmd)s: " + str(e), 'context': {'ctl_cmd': DYNAMIC_UI_CTL_CMD}}) from e if retcode != 0: raise UIConfigException({'message': "Error while executing %(ctl_cmd)s: " + out + " " + err, 'context': {'ctl_cmd': DYNAMIC_UI_CTL_CMD}}) def set_ui_config(parameters_dict): """ Sets UI settings to UI config file Using subfunction _set_ui_config :param parameters_dict: Parameters to set dictionary. For example: {'inodeLimits': {'showUserInodesUsage': False}, 'uiSettings': {'hideRubyApp': False, 'hideLVEUserStat': False, 'hidePythonApp': False, 'hidePHPextensions': True} } :return: None """ # Backward compatibility cl-config -> cloudlinux-selector on hidePythonApp try: # Try to read hidePythonApp key enablepythonapp = boolean_to_0_1(not parameters_dict['uiSettings']['hidePythonApp']) # Translate enablepythonapp binary key to disabled/enabled format for cl-selector lve_to_select = {'0':'disabled', '1': 'enabled'} # Call cloudlinux selector with translated param # CL-selector doesn't write changes to ui config file py_command = ("cloudlinux-selector set --json --interpreter python --selector-status " + lve_to_select[enablepythonapp]) subprocess.check_call(py_command, shell=True, executable='/bin/bash', stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT) except (subprocess.CalledProcessError, KeyError): pass # Call _set_ui_config with all params # (duplicate cl-selector call to write changes to config file) _set_ui_config(parameters_dict) def _set_ui_config(parameters_dict): """ Sets UI settings to config file :param parameters_dict: Parameters to set dictionary. For example: {'inodeLimits': {'showUserInodesUsage': False}, 'uiSettings': {'hideRubyApp': False, 'hideLVEUserStat': False, 'hidePythonApp': False, 'hidePHPextensions': True} } :return: None """ UIConfig().set_config(parameters_dict) # Synchronize params with according panel config file refresh_panel() def filter_flags(mapped_flags): panel_filter(mapped_flags.get('uiSettings')) try: if check_quota_enabled() is not None: mapped_flags.pop('inodeLimits') except KeyError: pass try: if not is_panel_feature_supported(Feature.LVE): mapped_flags.get('uiSettings', {}).pop('hideLVEUserStat') except KeyError: pass return mapped_flags def panel_filter(ui_settings): """ Function that filters UI Settings dictionary depending on panel type :param ui_settings: UI Settings that is filtered """ if not detect.is_cpanel() or \ not is_panel_feature_supported(Feature.RUBY_SELECTOR): ui_settings.pop('hideRubyApp') if detect.is_plesk() or not is_panel_feature_supported(Feature.PYTHON_SELECTOR): ui_settings.pop('hidePythonApp') if not is_panel_feature_supported(Feature.NODEJS_SELECTOR): ui_settings.pop('hideNodeJsApp') if not is_panel_feature_supported(Feature.XRAY): ui_settings.pop('hideXrayApp') if not is_panel_feature_supported(Feature.PHP_SELECTOR): ui_settings.pop('hidePhpApp') ui_settings.pop('hidePHPextensions')	4.74 KB	2026-02-17 10:40

Terminal