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-19 15:49
📄 __init__.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # from .mysql_governor_lib import ( MySQLGovernor, MySQLGovException, MySQLGovernorAbsent, GovernorStatus ) __all__ = [ "MySQLGovernor", "MySQLGovException", "MySQLGovernorAbsent", "GovernorStatus" ]	439 B	2026-02-10 15:10
📁 __pycache__	-	2026-03-19 15:49
📄 cledition.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import os import subprocess import warnings from enum import Enum from typing import AnyStr from jwt import exceptions from clcommon import clcagefs from clcommon.lib.consts import CLN_JWT_TOKEN_PATH, CL_EDITION_FILE_FOR_USERS from clcommon.lib.jwt_token import read_jwt, decode_jwt, jwt_token_check from clcommon.clexception import FormattedException # The function below used to be in this module, but were moved to clcommon.utils # So, importing them just for backward compatibility from clcommon.utils import get_os_version, is_ubuntu, is_secureboot_enabled # NOQA CL_SOLO_EDITION_FILE_MARKER = '/etc/cloudlinux-edition-solo' CL_ADMIN_EDITION_FILE_MARKER = '/etc/cloudlinux-edition-admin' SHARED_PRO_EDITION_HUMAN_READABLE = 'CloudLinux OS Shared Pro' SHARED_EDITION_HUMAN_READABLE = 'CloudLinux OS Shared' SOLO_EDITION_HUMAN_READABLE = 'CloudLinux OS Solo' ADMIN_EDITION_HUMAN_READABLE = 'CloudLinux OS Admin' class SupportedEditions(Enum): """ Keeps supported CloudLinux editions """ SOLO = 'solo' SHARED = 'shared' SHARED_PRO = 'shared_pro' ADMIN = 'admin' class CLEditionDetectionError(FormattedException): def __init__(self, message, context): FormattedException.__init__(self, { 'message': message, 'context': context }) HUMAN_READABLE_TOKEN_EDITION_PAIRS = { SOLO_EDITION_HUMAN_READABLE: SupportedEditions.SOLO.value, SHARED_PRO_EDITION_HUMAN_READABLE: SupportedEditions.SHARED_PRO.value, SHARED_EDITION_HUMAN_READABLE: SupportedEditions.SHARED.value, ADMIN_EDITION_HUMAN_READABLE: SupportedEditions.ADMIN.value } class CLEditions: @staticmethod def get_from_jwt(token=None, verify_exp=True): """ Note: be careful when modifying this method. Passing token in is used in X-Ray, ask @dkavchuk or someone else from C-Projects team for details :param token: :param verify_exp: :return: """ if token is None: token = read_jwt(CLN_JWT_TOKEN_PATH) try: jwt = decode_jwt(token, verify_exp=verify_exp) except exceptions.PyJWTError as e: raise CLEditionDetectionError( f'Unable to detect edition from jwt token: {CLN_JWT_TOKEN_PATH}. ' f'Please, make sure it is not broken, error: {e}' ) from e try: return jwt['edition'] except KeyError as e: # fallback for old format of tokens cl_plus = jwt.get('cl_plus') if cl_plus is None: raise CLEditionDetectionError( f'Unable to detect edition from jwt token: {CLN_JWT_TOKEN_PATH}. ' f'Please, make sure it is not broken, error: not enough fields for detection' ) from e return SupportedEditions.SHARED_PRO.value if cl_plus else SupportedEditions.SHARED.value @staticmethod def is_package_installed(package_name): with subprocess.Popen( ['rpm', '-q', package_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env={os.environ, {'LC_ALL': 'C'}}, ) as proc: out, err = proc.communicate() if proc.returncode != 0: if 'is not installed' in out: return False raise CLEditionDetectionError( f'Unable to check if package {package_name} is installed, ' f'reason: {out}, {err}' ) return True @classmethod def get_cl_edition( cls, skip_jwt_check=False, skip_marker_check=False, raw_jwt=None, verify_exp=True, ) -> str \| None: """ 1. Try to detect CL edition from JWT token; If edition field is in token -> return edition If edition field is not present -> recheck via cl_plus flag (see get_from_jwt) If token is broken -> raise an error about it 2. Try to detect edition from files. If JWT edition retreival was attempted and failed, or if we're in cagefs -> get edition from CL_EDITION_FILE_FOR_USERS If a edition marker file for Solo or Admin edition is present -> return corresponding edition Otherwise, fall back to CL_EDITION_FILE_FOR_USERS again Detection from marker files may be switched off using the `skip_marker_check` parameter. @param skip_jwt_check: if True, skip checking JWT token for edition. @param skip_marker_check: if True, skip checking marker files for edition. @param raw_jwt: raw JWT token to use for edition detection. If None, token will be read from file. @param verify_exp: if True, verify JWT token expiration time. @return: Edition of CloudLinux OS or None if edition could not be determined. Note: be careful when modifying this method. It is used in X-Ray, ask @dkavchuk or someone else from C-Projects team for details. """ # Skipping both jwt and file checks is not allowed if skip_jwt_check and skip_marker_check: raise CLEditionDetectionError( "Unable to detect edition: neither jwt token check, no file marker check enabled" ) # If we have a valid JWT token, try to get edition from it if not skip_jwt_check and os.path.isfile(CLN_JWT_TOKEN_PATH): try: edition = cls.get_from_jwt(token=raw_jwt, verify_exp=verify_exp) except PermissionError: edition = user_cl_edition() if edition: return edition # If user is in cagefs, look at the CL_EDITION_FILE_FOR_USERS file first if clcagefs.in_cagefs(): return user_cl_edition() # Otherwise, check for edition marker files if not skip_marker_check: # jwt has no 'edition' field -> ensure it really is Solo via file if os.path.isfile(CL_SOLO_EDITION_FILE_MARKER): return SupportedEditions.SOLO.value elif os.path.isfile(CL_ADMIN_EDITION_FILE_MARKER): return SupportedEditions.ADMIN.value # Finally, if no marker files are present, try CL_EDITION_FILE_FOR_USERS return user_cl_edition() # Couldn't detect edition through any means - return None return None def is_cl_solo_edition(skip_jwt_check=True, skip_marker_check=False, verify_exp=True): """ Allow skip_jwt_check ONLY if it not critical when license is not valid Use skip_marker_check=True when validity of license is critical and fallback to file is not applicable Note: be careful when modifying this method. It is used in X-Ray and SSA, ask @dkavchuk or someone else from C-Projects team for details """ edition = CLEditions.get_cl_edition( skip_jwt_check=skip_jwt_check, skip_marker_check=skip_marker_check, verify_exp=verify_exp ) return edition is not None and edition == SupportedEditions.SOLO.value def is_cl_shared_edition(skip_jwt_check=False, skip_marker_check=False, verify_exp=True): """ Allow skip_jwt_check ONLY if it not critical when license is not valid Use skip_marker_check=True when validity of license is critical and fallback to file is not applicable """ edition = CLEditions.get_cl_edition( skip_jwt_check=skip_jwt_check, skip_marker_check=skip_marker_check, verify_exp=verify_exp ) return edition is not None and edition == SupportedEditions.SHARED.value def is_cl_shared_pro_edition(skip_jwt_check=False, skip_marker_check=False, verify_exp=True): """ Allow skip_jwt_check ONLY if it not critical when license is not valid Use skip_marker_check=True when validity of license is critical and fallback to file is not applicable """ edition = CLEditions.get_cl_edition( skip_jwt_check=skip_jwt_check, skip_marker_check=skip_marker_check, verify_exp=verify_exp ) return edition is not None and edition == SupportedEditions.SHARED_PRO.value def is_cl_shared_pro_edition_safely(kwargs): """ Supress edition checking error and returns False """ try: return is_cl_shared_pro_edition(**kwargs) except CLEditionDetectionError: return False def is_cl_admin_edition(skip_jwt_check=False, skip_marker_check=False, verify_exp=True): """ Allow skip_jwt_check ONLY if it not critical when license is not valid Use skip_marker_check=True when validity of license is critical and fallback to file is not applicable """ edition = CLEditions.get_cl_edition( skip_jwt_check=skip_jwt_check, skip_marker_check=skip_marker_check, verify_exp=verify_exp ) return edition is not None and edition == SupportedEditions.ADMIN.value def get_cl_edition_readable() -> AnyStr: """ Function returns current edition of CL: - CloudLinux OS Shared - CloudLinux OS Shared Pro - CloudLinux OS Solo - Error string """ try: if is_cl_solo_edition(skip_jwt_check=True): return SOLO_EDITION_HUMAN_READABLE elif is_cl_admin_edition(skip_jwt_check=True): return ADMIN_EDITION_HUMAN_READABLE except CLEditionDetectionError: return SHARED_EDITION_HUMAN_READABLE success_flag, error_message, _ = jwt_token_check() # This error message indicates that user could not read jwt.token because not enough privileges # We have to check existence of token file, to be sure that privilege error acquired if 'read error' in error_message and (os.path.isfile(CLN_JWT_TOKEN_PATH) or clcagefs.in_cagefs()): with open(CL_EDITION_FILE_FOR_USERS, 'r', encoding='utf-8') as f: return f.read().strip() if success_flag: return SHARED_PRO_EDITION_HUMAN_READABLE else: return SHARED_EDITION_HUMAN_READABLE def print_skip_message_on_solo(): """Just print skip message""" warnings.warn( 'as part of unification effort, features might not map directly to editions', DeprecationWarning, ) print('CloudLinux Solo edition detected! \n' 'Command is skipped, because it is unsupported and unneeded on current edition') # NOTE(vlebedev): To properly avoid circular imports, these 2 functions' implementations # are moved to "utils" submodule of "clcommon.cpapi" as it is the level # where all necessary panel stuff is imported and constructed and which # uses "clcommon.lib" module as a dependency. In other words "clcommon.lib" lays # lower in the modules hierarchy and should not import from "clcommon.cpapi"). def skip_without_lve(): warnings.warn( 'since v3.4.1 use "clcommon.cpapi.utils:skip_without_lve" instead', DeprecationWarning, ) from clcommon.cpapi.utils import skip_without_lve # pylint: disable=cyclic-import,import-outside-toplevel return skip_without_lve() def lve_supported_or_exit(f): warnings.warn( 'since v3.4.1 use "clcommon.cpapi.utils:lve_supported_or_exit" instead', DeprecationWarning, ) from clcommon.cpapi.utils import lve_supported_or_exit # pylint: disable=cyclic-import,import-outside-toplevel return lve_supported_or_exit(f) def user_cl_edition(): """ This function is used as a workaround for users to be able to get the CL Edition. Crontab will write output of `cldetect --detect-edition` command to CL_EDITION_FILE_FOR_USERS, with 0644 permission. """ if os.path.exists(CL_EDITION_FILE_FOR_USERS): with open(CL_EDITION_FILE_FOR_USERS, "r", encoding="utf-8") as f: edition = f.read().strip() return HUMAN_READABLE_TOKEN_EDITION_PAIRS.get(edition) return None def is_container() -> bool: """ Determines is this system running inside container """ return os.path.exists('/etc/cloudlinux-container')	12.18 KB	2026-02-10 15:10
📄 cmt_utils.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import json import os import requests import subprocess import shutil from clcommon.lib.consts import ( DEFAULT_JWT_ES_TOKEN_PATH, PUSHGATEWAY_ADDRESS, DISABLE_CMT_FILE, ) def client_activation_data(raise_exception=False) -> dict: """ Check that client is activated. Absence of JWT token means that client can have volume license like GoDaddy In this case we return False. Also we return False in case of some errors: - Errors during request of status from CM - Malformed JWT token - Incorrect answer from CM """ try: with open(DEFAULT_JWT_ES_TOKEN_PATH, "r", encoding="utf-8") as f: token = f.read().strip() req = requests.get( f'{PUSHGATEWAY_ADDRESS}/api/cm/get_backend_activation', headers={"X-Auth": token}, timeout=25, ) req.raise_for_status() activation_info = req.json() except ( json.JSONDecodeError, requests.exceptions.RequestException, OSError, KeyError, IOError, ): if raise_exception: raise # we do not want to raise for cldiag return {} return activation_info def is_client_enabled(raise_exception=False) -> bool: return client_activation_data(raise_exception).get('activate', False) def collect_cmt_request_debug(cmt_domain): """ Several checks for CM - ping cmt domain - ping something that MUST be accessible - traceroute to CM """ process_results = [] ping_binary = shutil.which('ping') if ping_binary: ping_process = subprocess.run([ping_binary, '-c', '5', cmt_domain], check=False, capture_output=True, text=True) ping_external_process = subprocess.run([ping_binary, '-c', '5', 'google.com'], check=False, capture_output=True, text=True) ping_cloudlinux = subprocess.run([ping_binary, '-c', '5', 'cloudlinux.com'], check=False, capture_output=True, text=True) process_results.extend([ping_process, ping_cloudlinux, ping_external_process]) traceroute_binary = shutil.which('traceroute') if traceroute_binary: traceroute_process = subprocess.run([traceroute_binary, '-T', cmt_domain], check=False, capture_output=True, text=True) process_results.append(traceroute_process) return '\n'.join(f'Command: {process_data.args}\nRetcode: {process_data.returncode}\n' f'Stdout: {process_data.stdout}\nStderr: {process_data.stderr}\n' for process_data in process_results) def log_request_error_debug_info(component, logger, error, url, with_debug=False): """ Log debug info for request exceptions """ debug_info = 'N/A' if with_debug: debug_info = collect_cmt_request_debug(url) resolv_conf_info = get_resolv_conf_info() or 'N/A' logger.error('%s error while requesting %s: %s \n\n' 'DNS info: %s \n' 'Debug info for request: \n %s', component, url, str(error), str(resolv_conf_info), debug_info, extra={'fingerprint': ['{{ default }}', component, error.__class__.__name__, url]}) def get_resolv_conf_info(): """ Get info from revolv.conf """ resolv_conf = '/etc/resolv.conf' if not os.path.exists(resolv_conf): return None with open(resolv_conf, encoding='utf-8') as f: return f.read() def is_cmt_disabled() -> bool: """ Check that CM is disabled localy """ return os.path.exists(DISABLE_CMT_FILE)	4.17 KB	2026-02-10 15:10
📄 config.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import yaml import os class Config: CONFIG_PATH = '/etc/sysconfig/endservertools.yml' def __init__(self, config=CONFIG_PATH): self._config_path = config def _load(self): if not os.path.exists(self._config_path): return None try: with open(self._config_path, encoding='utf-8') as f: data = yaml.load(f, yaml.SafeLoader) except yaml.YAMLError: return None return data def get_value(self, parameter, default=None): data = self._load() if data is None: return default return data.get(parameter, default)	850 B	2026-02-10 15:10
📄 consts.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import os from clcommon.lib.config import Config CLN_JWT_TOKEN_PATH = '/etc/sysconfig/rhn/jwt.token' CL_EDITION_FILE_FOR_USERS = '/opt/cloudlinux/cl_edition' c = Config() DEFAULT_JWT_ES_TOKEN_PATH = c.get_value( 'jwt_token_file', CLN_JWT_TOKEN_PATH ) PUSHGATEWAY_ADDRESS = c.get_value( 'pushgateway_address', 'https://cm.cloudlinux.com' ) DEFAULT_PUSHGATEWAY_URL = f'{PUSHGATEWAY_ADDRESS}/pushgateway/store' DEFAULT_PUSHGATEWAY_CHECK_URL = f'{PUSHGATEWAY_ADDRESS}/pushgateway/check' DISABLE_CMT_DIR = "/etc/cl_plus" DISABLE_CMT_FILE = os.path.join(DISABLE_CMT_DIR, ".disabled") ACCELERATE_WP_INSTALLED_FROM_CM = os.path.join('/var/clwpos/admin', "awp_from_cm_installed.flag") ACCELERATE_WP_PREMIUM_INSTALLED_FROM_CM = os.path.join('/var/clwpos/admin', "awp_premium_from_cm_installed.flag")	1011 B	2026-02-10 15:10
📄 jwt_token.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # from jwt import decode, exceptions from clcommon.lib.consts import DEFAULT_JWT_ES_TOKEN_PATH def read_jwt(jwt_path): with open(jwt_path, mode='rb') as f: token_string = f.read().strip() return token_string def decode_jwt(token_string, verify_exp=True): jwt_token = decode(token_string, algorithms=['HS256'], options={'require_exp': True, "verify_exp": verify_exp, "verify_iss": True, 'verify_signature': False}, issuer='CloudLinux') if 'exp' not in jwt_token: raise exceptions.MissingRequiredClaimError('exp') return jwt_token def jwt_token_check(): """ JWT token check :return: Tuple: success_flag, error_message, token_string success_flag: True/False - OK, JWT is valid/Error, invalid error_message: Error message token_string: Token string """ success_flag, error_message, token_string = True, "OK", None try: token_string = read_jwt(DEFAULT_JWT_ES_TOKEN_PATH) except (OSError, IOError): return False, f"JWT file {DEFAULT_JWT_ES_TOKEN_PATH} read error", None # JWT read success try: jwt_token = decode_jwt(token_string) if jwt_token.get("cl_plus") is None: success_flag, error_message, token_string = False, "JWT token format error", None elif not jwt_token.get("cl_plus"): success_flag, error_message, token_string = False, "JWT token doesn't have CL+ service", None except exceptions.InvalidIssuerError: success_flag, error_message, token_string = False, "JWT token issuer is invalid", None except exceptions.ExpiredSignatureError: success_flag, error_message, token_string = False, "JWT token expired", None except exceptions.PyJWTError: success_flag, error_message, token_string = False, "JWT token format error", None return success_flag, error_message, token_string	2.1 KB	2026-02-10 15:10
📄 mysql_governor_lib.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # mysql_governor.py - module for interfacing with dbctl utility for get/set MySQL limits and user's status import os import xml.dom.minidom as xml import subprocess import json from xml.parsers.expat import ExpatError from typing import Tuple, Optional, Dict # NOQA from packaging.version import Version from clcommon.clexception import FormattedException from clcommon.utils_cmd import run_command, ExternalProgramFailed class GovernorStatus: ENABLED = 'enabled' ERROR = 'error' DISABLED = 'disabled' class MySQLGovException(FormattedException): pass class MySQLGovernorDisabled(MySQLGovException): """ Exception raised when dbgovernor daemon is offline """ def __init__(self): super().__init__({ 'message': "%(util)s is disabled in the system. " "Please, run \"%(command)s\" to start the service", 'context': { 'util': 'MySQL governor', 'command': 'service db_governor restart' } }) class MySQLGovernorAbsent(MySQLGovException): """ Exception raised when dbgovernor isn't installed """ def __init__(self): super().__init__({ 'message': "%(util)s not present in system", 'context': {'util': 'Governor'} }) def _get_exc_message(param): return { 'message': "Invalid %(param)s parameter", 'context': {'param': param} } class MySQLGovernor: """ MysqlGovernor library """ # Constants to calculate individial limits for cpu(current, short, middle period) # and read(current, short, middle, long periods) # cpu: # c, s = c95%, m = c87%, l = c75% # io (read == write): # c = io / 2, s = c83%, m = c76%, l = c59% # refer to LU-167 IO_PERCENTS = (0.83, 0.76, 0.59) # percents for calculate read/write limits CPU_PERCENTS = (0.95, 0.87, 0.75) # percents for calculate cpu limits _UTILITY_PATH = "/usr/sbin/dbctl" _PACKAGE_UTILITY_PATH = "/usr/share/lve/dbgovernor/governor_package_limitting.py" _GOVERNOR_BINARY_PATH = "/usr/sbin/db_governor" _CONTAINER_PATH = "/etc/container/mysql-governor.xml" # available for root only _CONTAINER_PATH_V2 = "/var/run/mysql-governor-config.xml" # available for everyone; governor-mysql >= 1.1-14 _S_DEFAULT = 'DEFAULT' def __init__(self): # List of ignored users in governor self._governor_ignored_users = None # MySQL Governor limits cache # Dictionary format: user_name -> (cpu_limit: int, io_limit: int, cpu_limit_marked: str, io_limit_marked: str) # cpu_limit: int, io_limit: int - "old" limits, without package differs marks. For backward compatibility # Both are integers always, IO limit - KB/s # cpu_limit_marked: str, io_limit_marked: str - "new" limits, with package differs marks. # Both are strings always, IO limit - KB/s # Examples: (150, 3072, '150', '3072'), (200, 3072, '200', '3072'), etc self._governor_limits = None # Error flag self._is_governor_error = False # MySQL Governor presence flag self._is_governor_present = os.path.isfile(self._UTILITY_PATH) self._governor_mode = self._detect_governor_mode() self._governor_version = self.get_governor_version() def get_governor_mode(self): return self._governor_mode def _detect_governor_mode(self): if self._is_governor_present: try: governor_cfg = self._get_xml_config() return governor_cfg.getElementsByTagName('lve')[0].getAttribute('use') except (MySQLGovException, IndexError): return None else: return None def _run_dbctl_with_args(self, args, check_exit_code=False): """ Run dbctl utility with given arguments and handle common errors: - governor is down: MySQLGovernorDisabled :param check_exit_code: whether we should raise exception when dbctl returs code != 0 """ ret_code, std_out, std_err = run_command( [self._UTILITY_PATH] + list(args), return_full_output=True) if "can't connect to socket" in std_out: # Governor not started self._is_governor_error = True self._governor_limits = None self._governor_ignored_users = None raise MySQLGovernorDisabled() elif check_exit_code and (ret_code or std_err): raise MySQLGovException({'message': "dbctl error: %(output)s", 'context': {'output': std_err}}) return ret_code, std_out, std_err def get_governor_status(self): # type: () -> Tuple[str, Optional[MySQLGovException]] return self._detect_governor_status() def _detect_governor_status(self): # type: () -> Tuple[str, Optional[MySQLGovException]] if self.is_governor_present(): try: self._load_info() except MySQLGovException as e: return GovernorStatus.ERROR, e except ExternalProgramFailed as e: return GovernorStatus.ERROR, MySQLGovException({ 'message': str(e) }) else: return GovernorStatus.ENABLED, None return GovernorStatus.DISABLED, None def _get_xml_config(self): # type: () -> xml.Document config_path = self._get_config_path() try: with open(config_path, 'r', encoding='utf-8') as f: return xml.parseString(f.read()) except (IOError, OSError) as e: self._is_governor_error = True raise MySQLGovException( {'message': "An error occured while loading governor " "config from %(path)s. Error: %(error)s", 'context': {'path': config_path, 'error': str(e)}} ) from e except ExpatError as e: self._is_governor_error = True raise MySQLGovException( {'message': "An error occured while parsing governor " "config from %(path)s. Error: %(error)s", 'context': {'path': config_path, 'error': str(e)}} ) from e @staticmethod def _load_ignore_users_from_xml(governor_cfg): """ Loads information about igrored users :type governor_cfg: xml.Document :return: list of ignore users """ ignore_users_list = [] try: gov_data = governor_cfg.getElementsByTagName("governor")[0] except IndexError as e: raise MySQLGovException({ 'message': 'Malformed mysql-governor config. ' 'Unable to find element \'%(element)s\'.', 'context': {'element': 'governor'} }) from e users_data_list = gov_data.getElementsByTagName("user") for user_data in users_data_list: user_mode = user_data.getAttribute("mode") if user_mode == "ignore": # ignore_users_list.append(user_data.getAttribute("mysql_name")) # ignore only system users if user_data.getAttribute("name"): ignore_users_list.append(user_data.getAttribute("name")) return ignore_users_list def _calc_rw_io_limits(self, io): """ Calculate db R/W IO limits based on governor version :param io: requested limits in MB/s :return: string values suitable to pass to "dbctl set ..." """ # Only Governor >= 1.2-18 has support for limits in bytes # This check should be removed when new Governor will be in stable repo if self._is_governor_newer_then('1.2-17'): io_limits = io 2 ** 20 # MB to Bytes limits_tmpl = "%sb,%sb,%sb,%sb" else: io_limits = io # Use MBytes as is if io_limits == 1: # This should prevent dropping to defaults even for old # Governor, and just set 1MB instead io_limits = 2 limits_tmpl = "%s,%s,%s,%s" read = limits_tmpl % self._percentage(int(io_limits // 2), self.IO_PERCENTS) write = limits_tmpl % self._percentage(int(io_limits // 2), self.IO_PERCENTS) return read, write def get_governor_version(self): if not self.is_governor_present(): return None try: res = subprocess.check_output([self._GOVERNOR_BINARY_PATH, '--version'], text=True) # example of valid output # res = 'governor-mysql version 1.2-36' version = res.strip().split(' ')[2] return version except (subprocess.CalledProcessError, OSError): return None def _is_governor_newer_then(self, version): current = self._governor_version if not current: return False # assume "No" if we can't determine version return Version(version) < Version(current) @staticmethod def _parse_line(line): """ Convert data line from dbctl to list :param line: Data line could be like: "default\t400/380/350/300\t953/791/724/562\t953/791/724/562" or: "default 400/380/350/300 1000/830/760/590 1000/830/760/590" depending on --kb/mb/bb option passed to dbctl :return: list: ['default', '400/380/350/300', '953/791/724/562', '953/791/724/562'] """ return [part for part in line.split() if part] @staticmethod def _percentage(value, percents): """ Calculate full list of governor limits by one value and percents koeff """ res = [value] for k in percents: res.append(int(valuek)) return tuple(res) def is_governor_present(self): """ Get governor presence flag :return: """ return self._is_governor_present def get_governor_status_by_username(self, username): """ Get MySQL governor status for supplied user :param username: Username for get status :return: Governor status: "watched"/"ignored" or None if error """ # Load Governor data self._load_info() if username in self._governor_ignored_users: return 'ignored' return 'watched' def get_limits_by_user(self, username, with_package_mark: bool = False) -> Tuple[int, int]: """ Get MySQL governor limits for supplied user :param username: Username for read limits :param with_package_mark: False - without package limits difference mark (for compatibility with non-package governor, used in cloudlinux-top/cloudlinux-statistics), True - with package limits difference mark :return: Tuple (CPU limit, IO limit). Examples: (150, 3072) - with_package_mark == False ('150', '3072') - with_package_mark == True ('150', '4096') - with_package_mark == True * - user has individual limits, differ from package limit MySQLGovException will be thrown if governor not present or error """ # Load Governor data if need self._load_info() if username in self._governor_limits: user_cpu_limit, user_io_limit_kb, user_cpu_limit_marked, user_io_limit_marked =\ self._governor_limits[username] else: user_cpu_limit, user_io_limit_kb, user_cpu_limit_marked, user_io_limit_marked = \ self._governor_limits['default'] if with_package_mark: limits_for_return = (user_cpu_limit_marked, user_io_limit_marked) else: limits_for_return = (user_cpu_limit, user_io_limit_kb) return limits_for_return def set_governor_status_for_user(self, username, status): """ Set MySQLGovernor status for single user :param: `str` username: Username for set status :param: `bool` status: True for "monitor", False for "ignore" :return: `bool`: operation status result """ self._load_info() status_cmd = "monitor" if status else "ignore" ret, std_out, std_err = self._run_dbctl_with_args([status_cmd, username]) if std_err or ret: exc_message = {'message': "Set governor status error(%(ret)s): %(output)s", 'context': {'ret': ret, 'output': std_err or std_out}} raise MySQLGovException(exc_message) return 0 def set_restricted_status_for_user(self, username, status): """ Set user restricted with dbctl utility :param: `str` username: Username for set restricted status :param: `bool` status: True for "restricted", False for "unrestricted" :return: `bool`: operation status result """ self._load_info() status_cmd = "restrict" if status else "unrestrict" if username in ["root", "admin"]: username = "default" ret, std_out, std_err = self._run_dbctl_with_args([status_cmd, username]) if std_err or ret: exc_message = {'message': "Set user restrict error(%(ret)s): %(output)s", 'context': {'ret': ret, 'output': std_err or std_out}} raise MySQLGovException(exc_message) return 0 def set_unrestricted_status_for_all_users(self): """ Set user restricted with dbctl utility :return: `bool`: operation status result """ self._load_info() ret, std_out, std_err = self._run_dbctl_with_args(["unrestrict-all"]) if std_err or ret: exc_message = {'message': "Set all users unrestrict status error(%(ret)s): %(output)s", 'context': {'ret': ret, 'output': std_err or std_out}} raise MySQLGovException(exc_message) return 0 def get_restrict_status_by_username(self, username): """ Get MySQL governor status for supplied user :param username: Username for get status :return: Governor restricted status: "restricted"/"unrestricted" """ # Load Governor data self._load_info() if username in ["root", "admin"]: username = "default" if username in self._governor_restricted_users: return 'restricted' return 'unrestricted' def set_limits_for_user(self, username, cpu=None, io=None): # this function interface for full edit mode # def set_limits_for_user(self, username, cpu=None, io=None, read=None, # write=None): """ Set MySQLGovernor limits for user :param: username `str`: username for set limits :param: `int`\|`list` cpu: governor cpu limit. when it param int - calculate by percentage other params :param: `int`\|`list` io: io value means that read and write limits similar :param: `int`\|`list` read: read limit :param: `int`\|`list` write: write limit :return: 0 """ if cpu is None and io is None: # and read is None and write is None: return 0 self._load_info() cmd = ["set", username] if cpu is not None: if isinstance(cpu, int): cpu = ",".join(map(str, self._percentage(cpu, self.CPU_PERCENTS))) else: raise MySQLGovException(_get_exc_message('cpu')) # uncomment this lines for add full edit mode # elif isinstance(cpu, (list, tuple)) and len(cpu) == 4: # cpu = ",".join(map(str, cpu)) # else: # raise MySQLGovException(_get_exc_message('cpu')) cmd.append(f"--cpu={cpu}") if io is not None: # uncomment this line for add full edit mode # if io is not None or read is not None or write is not None: if isinstance(io, int): read, write = self._calc_rw_io_limits(io) else: raise MySQLGovException(_get_exc_message('io')) # uncomment this lines for add full edit mode # elif isinstance(io, (list, tuple)) and len(io) == 4: # read = write = ",".join(map(str, io)) # else: # if isinstance(read, int): # read = "%s,%s,%s,%s" % (read, read0.83, read0.76, read0.59) # elif isinstance(read, (list, tuple)) and len(read) == 4: # read = ",".join(map(str, read)) # if isinstance(write, int): # write = "%s,%s,%s,%s" % (write, write0.83, write0.76, write0.59) # elif isinstance(write, (list, tuple)) and len(write) == 4: # write = ",".join(map(str, write)) # else: # raise MySQLGovException(_get_exc_message('limit')) cmd.append(f"--read={read}") cmd.append(f"--write={write}") try: ret, std_out, std_err = self._run_dbctl_with_args(cmd) except ExternalProgramFailed as e: raise MySQLGovException(str(e)) from e if std_err or ret: exc_message = {'message': "Set all users unrestrict status error(%(ret)s): %(output)s", 'context': {'ret': ret, 'output': std_err or std_out}} raise MySQLGovException(exc_message) # Reset users limits cache self._governor_limits = None return 0 def _get_package_raw_limits_from_utility(self, package_name: Optional[str]) -> dict: """ Retrieve MySQL Governor package limits :param package_name: Package name. If None, get all packages name :return: Dict with limits. Example: {'pack1': {'cpu': 100, 'io': 900}, 'pack2': {'cpu': 100, 'io': 900}} """ if package_name: cmd_list = [self._PACKAGE_UTILITY_PATH, 'get', f"--package={package_name.encode().decode('unicode-escape')}", '--format=kb'] else: cmd_list = [self._PACKAGE_UTILITY_PATH, 'get', '--all', '--format=kb'] ret_code, std_out, std_err = run_command(cmd_list, return_full_output=True) if ret_code != 0 or std_err: raise MySQLGovException({ 'message': "'%(cmd)s' failed, stderr is: %(stderr)s", 'context': {'cmd': ' '.join(cmd_list), 'stderr': std_err} }) try: package_data_from_util = json.loads(std_out) except (json.JSONDecodeError, ) as e: raise MySQLGovException({ 'message': "%(util)s output invalid, error is: %(error)s", 'context': {'util': self._PACKAGE_UTILITY_PATH, 'error': str(e)} }) from e return package_data_from_util def _calc_package_limits_from_raw(self, cpu_limits_list: list, read_limits_list: list, write_limits_list: list) -> Tuple[int, int]: """ Calculate package limits from raw governor limits :param cpu_limits_list: CPU limits list :param read_limits_list: Read limits list :param write_limits_list: Write limits list :return: Tuple: (cpu_limit, io_limit) """ return int(cpu_limits_list[0]), self._get_user_io_limit(str(read_limits_list[0]), str(write_limits_list[0])) def get_package_limits(self, package_name: Optional[str] = None) -> Optional[Dict]: """ Retrieve MySQL Governor package limits :param package_name: Package name. If None, get all packages name :return: Dict with limits. Example: {'pack1': {'cpu': 100, 'io': 900}, 'pack2': {'cpu': 100, 'io': 900}} """ try: package_data_from_util = self._get_package_raw_limits_from_utility(package_name) except MySQLGovException: # pylint: disable=try-except-raise raise # Convert limits from governor packages_data = {} for pack_name, pack_limits in package_data_from_util.items(): cpu_limit, io_limit = self._calc_package_limits_from_raw(pack_limits['cpu'], pack_limits['read'], pack_limits['write']) packages_data[pack_name] = {'cpu': cpu_limit, 'io': io_limit} return packages_data def reset_user_limits_to_defaults(self, username: str, limit_names: list): """ Reset users limits to default :param username: User name ro reset limits :param limit_names: Limit names list to reset """ # /usr/share/lve/dbgovernor/governor_package_limitting.py reset_individual \ # --user=res1 --limits=mysql-cpu,mysql-io limits_string = ','.join(limit_names) cmd_list = [self._PACKAGE_UTILITY_PATH, 'reset_individual', f'--user={username}', f'--limits={limits_string}'] ret_code, _, std_err = run_command(cmd_list, return_full_output=True) # For reliability we check both retcode and std_err if ret_code != 0 or std_err: raise MySQLGovException({ 'message': "'%(cmd)s' is failed, stderr is: %(stderr)s", 'context': {'cmd': ' '.join(cmd_list), 'stderr': std_err} }) def _set_governor_limits_to_cpanel_package( # pylint: disable=too-many-branches self, package_name: str, cpu_limit: Optional[int], io_limit: Optional[int] ): """ Set MySQL Governor to cPanel package file. If limits not changed, package file will not be written :param package_name: Package name :param cpu_limit: MySQL CPU limit to set :param io_limit: MySQL IO limit to set """ from clcommon.utils import get_file_lines, write_file_lines # pylint: disable=import-outside-toplevel package_path = f'/var/cpanel/packages/{package_name}' try: cpanel_package_lines = get_file_lines(package_path) except (OSError, IOError, ): return if len(cpanel_package_lines) == 0: return lines_to_write = [] is_change_made = False # Find and change limit lines: # lve_mysql_cpu=4000 # lve_mysql_io=4096 for line in cpanel_package_lines: if line.startswith('lve_mysql_cpu') and cpu_limit is not None: parts = line.strip().split('=') if len(parts) != 2: continue if cpu_limit == 0: s_cpu_limit = self._S_DEFAULT else: s_cpu_limit = str(cpu_limit) s_old_cpu_limit = parts[1].strip() if s_old_cpu_limit != s_cpu_limit: lines_to_write.append(f'lve_mysql_cpu={s_cpu_limit}\n') is_change_made = True else: # MYSQL CPU limit unchanged, save old line lines_to_write.append(f'{line}\n') elif line.startswith('lve_mysql_io') and io_limit is not None: if io_limit == 0: s_io_limit = self._S_DEFAULT else: s_io_limit = str(io_limit) parts = line.strip().split('=') if len(parts) != 2: continue value = parts[1].strip() if value != s_io_limit: lines_to_write.append(f'lve_mysql_io={s_io_limit}\n') is_change_made = True else: # MYSQL IO limit unchanged, save old line lines_to_write.append(f'{line}\n') else: lines_to_write.append(line) if is_change_made: write_file_lines(package_path, lines_to_write, 'w') def _apply_package_limits_to_cpanel(self, package_name: str, cpu_limit: Optional[int], io_limit: Optional[int]): """ Apply all MySQL Governor packages limits to cpanel package file. In not cPanel, do nothing :param package_name: Package name to update :param cpu_limit: MySQL CPU limit to set, None - not change :param io_limit: MySQL IO limit to set, None - not change """ from clcommon.cpapi import getCPName # pylint: disable=import-outside-toplevel if getCPName() != 'cPanel': return self._set_governor_limits_to_cpanel_package(package_name, cpu_limit, io_limit) def set_package_limits(self, package_name: str, cpu_limit: Optional[int] = None, io_limit: Optional[int] = None): """ Set limits for Governor package :param package_name: Package name for set :param cpu_limit: MySQL CPU limit to set :param io_limit: MySQL CPU limit to set """ # Argument validation # /usr/share/lve/dbgovernor/governor_package_limitting.py set --package pack2 --cpu=200,201,202,203 # --read=500,501,502,503 --write=400,401,402,403 if package_name is None or cpu_limit is None and io_limit is None: raise MySQLGovException("MySQLGovernor.set_package_limits arguments error: " "Package name and at least one limit " f"should be provided. Current arguments: package name: {package_name}; " f"cpu limit is {cpu_limit}; IO limit is {io_limit};") cmd_list = [self._PACKAGE_UTILITY_PATH, 'set', '--package', package_name] # Check arguments and prepare command line parameters for governor utility if cpu_limit is not None: if isinstance(cpu_limit, int): cpu = ",".join(map(str, self._percentage(cpu_limit, self.CPU_PERCENTS))) else: raise MySQLGovException(_get_exc_message('cpu_limit')) cmd_list.append(f"--cpu={cpu}") if io_limit is not None: if isinstance(io_limit, int): read, write = self._calc_rw_io_limits(io_limit) else: raise MySQLGovException(_get_exc_message('io_limit')) cmd_list.append(f"--read={read}") cmd_list.append(f"--write={write}") ret_code, std_out, std_err = run_command(cmd_list, return_full_output=True) if ret_code != 0 or std_err: raise MySQLGovException({ 'message': "'%(command)s' is failed, stdout is: '%(stdout)s', stderr is: '%(stderr)s'", 'context': {'command': ' '.join(cmd_list), 'stdout': std_out, 'stderr': std_err} }) # Apply limits to cPanel package self._apply_package_limits_to_cpanel(package_name, cpu_limit, io_limit) def _get_config_path(self): """ Get config path for mysql-governor; :rtype: str\|None """ if os.path.isfile(self._CONTAINER_PATH_V2): return self._CONTAINER_PATH_V2 return self._CONTAINER_PATH def _read_ignore_users(self): """Load ignore users list from container file""" try: governor_xml = self._get_xml_config() self._governor_ignored_users = \ self._load_ignore_users_from_xml(governor_xml) except MySQLGovException: self._governor_limits = None self._governor_ignored_users = None raise def _load_info(self): """ Loads users info from MySQL governor :return: None """ # Exit if governor data already loaded if self._governor_ignored_users is not None and self._governor_limits is not None: return # Exit if governor not present if not self._is_governor_present: self._is_governor_error = True raise MySQLGovernorAbsent() utility_exc_message = {'message': "%(utility)s output is invalid", 'context': {'utility': self._UTILITY_PATH}} self._read_ignore_users() # Load governor limits is_kb_limits_ok, gov_data_str = self._run_dbctl_list() _, gov_restricted_str, _ = \ self._run_dbctl_with_args(['list-restricted'], check_exit_code=True) self._governor_restricted_users = [ line.split()[0] for line in gov_restricted_str.strip().split('\n')[1:] ] # Parse dbctl output gov_data_lines = gov_data_str.split('\n') self._governor_limits = self._parse_dbctl_data_lines(gov_data_lines, is_kb_limits_ok, utility_exc_message) # Check default settings presence if 'default' not in self._governor_limits: self._is_governor_error = True self._governor_limits = None self._governor_ignored_users = None exc_message = { 'message': "There is no %(what)s found in %(where)s", 'context': {'what': 'default settings', 'where': f'{self._UTILITY_PATH} output'} } raise MySQLGovException(exc_message) @staticmethod def _get_user_io_limit(read_limit: str, write_limit: str): """ Calculates the io limit. Handles the situation when user's write or read io limit is less than 1mb/s (PTCLLIB-85). :type write_limit: str :type read_limit: str :rtype: int """ try: user_io_limit = int(read_limit) + int(write_limit) except ValueError: if read_limit == write_limit == "<1": user_io_limit = 1 elif write_limit == "<1": user_io_limit = int(read_limit) else: user_io_limit = int(write_limit) return user_io_limit def _run_dbctl_list(self, _is_incorrect_syntax=False): """ Executes dbctl list-marked --kb or dbctl list-marked :param _is_incorrect_syntax: True is emulate dbctl error. Only for testing! :return: Cortege (is_kb_limits_ok, stdout_str), where is_kb_limits_ok == True, if dbctl returned limits in KB, else - False stdout_str - dbctl stdout string """ ret_code, gov_data_str, _ = self._run_dbctl_with_args( ['list-marked', '--kb'], check_exit_code=True) # Check is KB limits supported is_kb_limits_ok = True if _is_incorrect_syntax or 'Incorrect syntax' in gov_data_str: # --kb option not supported, call without it _, gov_data_str, _ = self._run_dbctl_with_args( ['list'], check_exit_code=True) is_kb_limits_ok = False return is_kb_limits_ok, gov_data_str def _parse_dbctl_data_lines(self, data_lines_list, is_kb_limits_ok: bool, utility_exc_message: dict) -> dict: """ Converts data lines from dbctl stdout to dictionary :param data_lines_list: List of lines from dbctl stdout :param is_kb_limits_ok: Is limits already in KB/s :param utility_exc_message: Message dict for exception :return: Tuple(dict, dict) dbctl data dictionary. Example: {'default': (400, 1953124, '400', '1953124'), 'cltest1': (350, 2025138, '350', '2025138') } """ governor_limits = {} for line in data_lines_list: line = line.strip() # Pass header line and empty lines if not line or 'cpu(%)' in line: continue # List: [0] - username, [1] - CPU limits, [2] - read limits, [3] - write limits, [4] - package limits marks user_limits_list = MySQLGovernor._parse_line(line) # Data format verification if len(user_limits_list) != 5 or len(user_limits_list[4]) != 2: self._is_governor_error = True self._governor_limits = None self._governor_ignored_users = None raise MySQLGovException(utility_exc_message) cpu_limits_list = user_limits_list[1].split('/') # '400/380/350/300' read_limits_list = user_limits_list[2].split('/') # '1000/830/760/590' write_limits_list = user_limits_list[3].split('/') # '1000/830/760/590' if len(cpu_limits_list) != 4 or len(read_limits_list) != 4 or len(write_limits_list) != 4: self._is_governor_error = True self._governor_limits = None self._governor_ignored_users = None raise MySQLGovException(utility_exc_message) user_name = user_limits_list[0] # Determine CPU limit as [0] limit user_cpu_limit = int(cpu_limits_list[0]) # Determine IO limit as read_limit[0] + write_limit[0] user_io_limit = self._get_user_io_limit(read_limits_list[0], write_limits_list[0]) # limit if is_kb_limits_ok else limit1024 user_io_limit_kb = user_io_limit if is_kb_limits_ok else user_io_limit1024 # Process package differ marks # Package limits marks are placed in user_limits_list[4] and shows as '+-' (for example) # There are 2 marks always, each or '-' or '+' # 1st mark - CPU limit # 2nd mark - IO limit # If mark is '-' - package limit, if '+' - individual (should be marked by '') marks = user_limits_list[4] if marks[0] == '+': user_cpu_limit_marked = f'{cpu_limits_list[0]}' else: user_cpu_limit_marked = cpu_limits_list[0] if marks[1] == '+': user_io_limit_marked = f'{user_io_limit_kb}' else: user_io_limit_marked = str(user_io_limit_kb) # Add limits to dictionary governor_limits[user_name] = (user_cpu_limit, user_io_limit_kb, user_cpu_limit_marked, user_io_limit_marked) return governor_limits	33.99 KB	2026-02-10 15:10
📄 network.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import socket def get_hostname(): try: return socket.gethostname() except Exception: return None def get_ip_addr(hostname): try: return socket.gethostbyname(hostname) except socket.error: return None	452 B	2026-02-10 15:10
📄 whmapi_lib.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # """ Everything that is related to whmapi calls """ import json from clcommon import FormattedException from clcommon.utils import run_command from urllib.parse import urlencode __all__ = ('WhmApiRequest', 'WhmApiError') class WhmApiError(FormattedException): """ An error that is raised in case of an error in communication with whmapi. """ def __init__(self, message, context): FormattedException.__init__(self, { 'message': message, 'context': context }) class WhmLicenseError(WhmApiError): """A license-related error raised by whmapi.""" pass class WhmNoPhpBinariesError(WhmApiError): """ An error when there are no installed php binaries """ pass class WhmApiRequest: """ Wrapper over cpanel's whm command-line api tool that allows us to easily build complex requests (filter, sorting, etc) See details in the official cpanel docs (link below) https://documentation.cpanel.net/display/DD/Guide+to+WHM+API+1 """ WHMAPI = '/usr/sbin/whmapi1' API_RESULT_OK = 1 def __init__(self, command): self._command = command self._filters = {} self._args = {} self._extra_args = ['--output', 'json'] def _run_whmapi(self, command): exitcode, output, _ = run_command( command, return_full_output=True) if exitcode != 0: raise WhmApiError( 'whmapi exited with code %(code)i', code=exitcode ) try: response = json.loads(output) # TODO: PTCCLIB-196 # Starting with cPanel v86.0 whmapi1 returns empty reason # if we try to revome nonexistent package. # It's temporary solution until cPanel provides another one. if ('metadata' in response)\ and ('reason' in response['metadata'])\ and (response['metadata']['reason'] is None): response['metadata']['reason'] = '' except (TypeError, ValueError) as e: raise WhmApiError( 'whmapi returned invalid response that ' 'cannot be parsed with json, output: %(output)s', output=output ) from e self._validate(response) return response @classmethod def _validate(cls, response): """ Check response metadata """ if cls._is_license_error(response): raise WhmLicenseError( 'whmapi license error: %(response)s', response=response['statusmsg'] ) if cls._is_no_php_binaries_error(response): raise WhmNoPhpBinariesError( 'Php binaries error: %(response)s', response=response['metadata']['reason'] ) try: result, reason = \ response['metadata']['result'], response['metadata']['reason'] except KeyError as e: raise WhmApiError( 'whmapi metadata section is broken, output: %(response)s', response=response ) from e # in 'ideal' world this should never happen as we check whmapi exitcode if result != WhmApiRequest.API_RESULT_OK: raise WhmApiError( 'whmapi failed to execute request, reason: %(reason)s', reason=reason ) @staticmethod def _is_license_error(response): """ Distinguish license-related WHM API errors from others. License errors are on the client's side, and should not be logged to sentry. An error is considered license-related if the API returns status 0 and the error message contains the word 'license' """ return ('statusmsg' in response and response['status'] == 0 and 'license' in response['statusmsg'].lower()) @staticmethod def _is_no_php_binaries_error(response): """ No binaries error can be detected by checking special message '“PHP” is not installed on the system' whmapi output """ return ('metadata' in response and 'reason' in response['metadata'] and '“PHP” is not installed on the system' in response['metadata']['reason']) def with_arguments(self, kwargs): """ Add some extra arguments to subprocess call Useful for methods like createacct, removeacct :param kwargs: arguments that will be added to cmd :rtype: WhmApiRequest """ self._args.update(kwargs) return self # TODO: enable in the future and add some unittests # def filter(self, **kwargs): # """ # Implements output filtering, see the following url for details # https://documentation.cpanel.net/display/DD/WHM+API+1+-+Filter+Output # :param kwargs: dict # """ # self._filters.update(kwargs) # return self def call(self): """ Run subprocess, run output validation and return json-loaded response :return: """ cmd = [ self.WHMAPI, self._command ] for k, v in list(self._args.items()): # https://documentation.cpanel.net/display/DD/Guide+to+WHM+API+1 if isinstance(v, bool): # the term "boolean" in our documentation refers # to parameters that accept values of 1 or 0. # cPanel & WHM's APIs do not support the literal # values of true and false. v = int(v) argument = urlencode({k: v}) cmd.append(argument) # TODO: enable in the future and add some unittests # for key, value in self._filters.items(): # cmd.extend('api.filter.{}={}'.format(key, value)) cmd.extend(self._extra_args) result = self._run_whmapi(cmd) if 'data' in result: # for getting method return result['data'] elif 'output' in result: # for setting method return result['output'] else: return result	6.3 KB	2026-02-10 15:10

Terminal