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 import subprocess def exec_utility(util_path, params, stderr=False): """ Executes supplied utility with supplied parameters :param util_path: string with path to callable utility :param params: utility parameters :param stderr: trigger to return stderr or not :return: Cortege (ret_code, cagefsctl_stdout) if stderr is False Cortege (ret_code, cagefsctl_stdout, cagefsctl_stderr) otherwise """ args = [] args.append(util_path) args.extend(params) with subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: out, err = proc.communicate() retcode = proc.returncode if stderr: return retcode, out.strip(), err.strip() return retcode, out.strip()	1015 B	2026-02-17 10:40
📁 __pycache__	-	2026-03-27 02:38
📄 arg_parsers.py # coding:utf-8 # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from typing import Tuple, Union from docopt import DocoptExit, docopt from schema import Schema, And, Use, Or, SchemaError from clcommon.const import Feature from clcommon.cpapi import is_panel_feature_supported from .utils import VALUES_STR PROG_NAME = "cloudlinux-limits" AVAILABLE_LVE_KEYS = {"speed", "nproc", "pmem", "vmem", "maxEntryProcs", "io", "iops"} AVAILABLE_LVP_KEYS = {"speed", "nproc", "pmem", "maxEntryProcs", "io", "iops"} AVAILABLE_QUOTA_KEYS = {"inodes"} AVAILABLE_MYSQL_KEYS = {"mysql-cpu", "mysql-io"} AVAILABLE_KEYS = AVAILABLE_LVE_KEYS \| AVAILABLE_QUOTA_KEYS \| AVAILABLE_MYSQL_KEYS \| AVAILABLE_LVP_KEYS AVAILABLE_DEFAULTS = AVAILABLE_KEYS \| {"all"} if is_panel_feature_supported(Feature.LVE): AVAILABLE_LIMITS = AVAILABLE_KEYS \| {"mysql-gov", "mysql-restrict", "cagefs"} else: AVAILABLE_LIMITS = AVAILABLE_QUOTA_KEYS \| {"cagefs"} AVAILABLE_MYSQL_KEYS_ALL = AVAILABLE_MYSQL_KEYS \| {"mysql-restrict", "mysql-unrestrict-all", "mysql-gov"} ADMIN_ONLY_OPTIONS = ["--mysql-restrict", "--mysql-unrestrict-all", "--mysql-gov", "--mysql-io", "--mysql-cpu", "--cagefs", "--inodes"] def _default_keys_validate(keys): """ Validate limits keys for --default """ return len(set(keys) - AVAILABLE_DEFAULTS) == 0 def _limits_keys_validate(keys): """ Validate limits keys """ return len(set(keys.split(",")) - set(AVAILABLE_LIMITS)) == 0 # 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))) global_schema = Schema({ "set": bool, "get": bool, "disable-reseller-limits": bool, "enable-reseller-limits": bool, "--pmem-lower-bound": Or(None, str), "--json": And(bool, lambda x: x, error="use --json option, other modes currently unsupported"), "--human-readable-numbers": bool, "--unlimited": bool, "--default": Or(None, And(Use(lambda x: x.split(',')), _default_keys_validate), error="Invalid keys"), "--save-all-parameters": bool, "--lve-id": Or(None, And(Use(int), lambda x: int(x) >= 0), And(Use(lambda x: str(x).lower()), lambda x: x == "default"), error="--lve-id must be non-negative integer value or 'default'"), "--reseller-id": Or(None, And(Use(int), lambda x: int(x) >= 0), error="--reseller-id must be non-negative integer value"), "--reseller-name": Or(None, And(str, lambda x: not x.isdigit()), error="Invalid reseller name"), "--all": bool, "--for-reseller": Or(None, And(str, lambda x: not x.isdigit()), error="Invalid reseller name"), "--username": Or(None, str), "--domain": Or(None, str), "--limits": Or(None, _limits_keys_validate, error="Invalid keys"), "--mysql-gov": Or(None, lambda x: x in ["ignored", "watched"], error=f"{VALUES_STR}: 'ignored', 'watched'"), "--cagefs": Or(None, lambda x: x in ["enabled", "disabled"], error=f"{VALUES_STR}: 'enabled', 'disabled'"), "--mysql-restrict": Or(None, lambda x: x in ["restricted", "unrestricted"], error=f"{VALUES_STR}: 'restricted', 'unrestricted'"), "--mysql-unrestrict-all": bool, "--speed": Or(None, str), "--pmem": Or(None, str), "--vmem": Or(None, str), "--nproc": Or(None, str), "--io": Or(None, str), "--iops": Or(None, str), "--maxEntryProcs": Or(None, str), "--mysql-cpu": Or(None, And(Use(int), lambda x: x >= 0), error="--mysql-cpu must be non-negative integer value"), # uncomment this for add full edit mode # "--mysql-io": Or(None, And(Use(int), lambda x:x >= 0), gov_int_list, "--mysql-io": Or(None, And(Use(int), lambda x: x >= 0), error="--mysql-io must be non-negative integer value"), # uncomment this for add full edit mode # "--mysql-read": Or(None, And(Use(int), lambda x:x >= 0), gov_int_list, # error="--mysql-read must be non-negative integer value or 4 int numbers with comma"), # "--mysql-write": Or(None, And(Use(int), lambda x:x >= 0), gov_int_list, # error="--mysql-write must be non-negative integer value or 4 int numbers with comma"), "--inodes": Or(None, str), "--help": bool, }) USAGE_CMDS = ( "set [--json] (--lve-id <int> \| --username <str>) (--unlimited) [--for-reseller <str>]", "set [--json] (--lve-id <int> \| --username <str> \| --reseller-name <str>) (--default <str>)", "set [--json] (--lve-id <int> \| --username <str>) " "(--mysql-gov <ignored,watched> \| --cagefs <enabled,disabled> \| --mysql-restrict <restricted,unrestricted>)", "set [--json] (--mysql-unrestrict-all)", "set [--json] (--lve-id <int> \| --username <str>) " "[--speed <str> --io <str> --nproc <str> --pmem <str> --vmem <str> --iops <str> --inodes <N,M> " "--maxEntryProcs <str> --mysql-cpu <int> --mysql-io <int> --save-all-parameters]", "set [--json] (--lve-id <int> \| --username <str> \| --reseller-name <str>) " "[--speed <str> --io <str> --nproc <str> --pmem <str> --vmem <str> --iops <str> --maxEntryProcs <str>] " "[--default <str>] [--for-reseller <str>]", "set [--json] (--reseller-name <str>) (--unlimited)", "set [--json] (--pmem-lower-bound=<str>)", "[get] [--json] [--lve-id <int> \| --username <str> \| --reseller-name <str> \| --domain <str>] " "[--limits=<keys>] [--human-readable-numbers] [--for-reseller <str>]", "disable-reseller-limits (--reseller-id <int> \| --reseller-name <str>) [--json]", "enable-reseller-limits ((--reseller-id <int> \| --reseller-name <str>) \| --all) [--json]", "(-h \| --help)", ) USAGE = "\n ".join(f"{PROG_NAME} {cmd}" for cmd in USAGE_CMDS).strip() class CloudlinuxLimitsOptsParser: """ Parse arguments for cloudlinux-limits command :param argv: sys.argv :param _is_json_need: sys.argv contains --json key :return cortege: (error_flag, s_message) """ docstring = f"""Utility to get/set any Cloudlinux limits Usage: {USAGE} Options: --json Return data in JSON format. --lve-id <int> LVE id. will display record only for that LVE id. --username <str> Execute command only for specific user. --reseller-name <str> Execute command only for specific reseller. --reseller-id <int> Execute command only for specific reseller. --all Execute command for all resellers --for-reseller <str> Use supplied reseller for get/set data. --domain <str> Show data only for specific domain --limits <keys> Available keys: speed,nproc,pmem,vmem,maxEntryProcs,io, iops,mysql-gov,mysql-cpu,mysql-io,cagefs,inodes --human-readable-numbers Return PMEM and VMEM limits in KBytes, MBytes or GBytes --unlimited Set all limits to unlimited. --default [limits] Reset limits to the Package defaults. List of comma-separated limits to reset them to default or "all" --mysql-gov <ignored\|watched> Monitor or ignore by MySQL governor. --cagefs <enabled\|disabled> Enable or disable CageFS for a user. --mysql-restrict <[un]restricted> Set user restrict status with dbctl (restricted or unrestricted). --mysql-unrestrict-all Unrestrict all restricted users with dbctl. --speed <str> Limit CPU usage for LVE \| LVP. --pmem <str> Limit physical memory usage for applications inside LVE \| LVP. --vmem <str> Limit virtual memory for applications inside LVE. --nproc <str> Limit number of processes for LVE \| LVP. --io <str> Define io limits for LVE \| LVP (KB/s). --iops <str> Limit io per second for LVE \| LVP. --maxEntryProcs <str> Limit number of entry processes for LVE \| LVP. --mysql-cpu <int> Set MySQL governor CPU limit (pct). --mysql-io <int> Set MySQL governor IO limit (read + write MB/s) --inodes <N,M> Set inode limits. N - soft, M - hard. --pmem-lower-bound <str> Set the PMEM lower bound for the limits. --save-all-parameters Save all parameters even if they match with defaults settings. -h, --help Show this help message and exit """ # add this parameters to docstring when need add read/write/cpu full edit mode # this line for Usage block # {0} set [--json] (--lve-id <int> \| --username <str>) \ # [--speed <int> --io <int> --nproc <int> --pmem <int> --vmem <int> --iops <int> --inodes <N,M> \ # --maxEntryProcs <int> --mysql-cpu <int:int,int,int,int> --mysql-read <int:int,int,int,int> \ # --mysql-write <int:int,int,int,int> --mysql-io <int:int,int,int,int> --save-all-parameters] # this lines for Options block # --mysql-cpu <int:int,int,int,int> Set MySQL governor CPU limit (pct). # --mysql-io <int:int,int,int,int> Set MySQL governor IO limit (read + write KB/s) # --mysql-read <int:int,int,int,int> Set MySQL governor read limit (KB/s) # --mysql-write <int:int,int,int,int> Set MySQL governor write limit (KB/s) commands = ("set", "get", "disable-reseller-limits", "enable-reseller-limits") @staticmethod def _is_option_compatible_with_admins(option: str, argv: list) -> Tuple[bool, str]: """ Checks options compatibility with admin options :return: Cortege (flag, message) flag: True/False - compatible/not compatible message - comment if flag is False """ option_found = False admin_option_name = "" for arg in argv: if arg.startswith(option): option_found = True continue if not admin_option_name: for admin_option in ADMIN_ONLY_OPTIONS: if arg.startswith(admin_option): admin_option_name = admin_option if option_found and admin_option_name: return False, f"ERROR: option '{admin_option_name}' is not compatible with '--for-reseller'" return True, "" def _perform_checks_before_docopt(self, argv: list) -> Tuple[bool, str]: """ Performs checks before parsing arguments """ is_compatible, message = self._is_option_compatible_with_admins("--for-reseller", argv) return is_compatible, message def _perform_checks_after_docopt(self, args: dict) -> Tuple[bool, str]: """ Performs checks after parsing arguments """ # cloudlinux-limits set/get --json --reseller-name=reseller --for-reseller=reseller -- ERROR if args["--reseller-name"] and args["--for-reseller"]: return False, "ERROR: '--reseller-name' and '--for-reseller' cannot be used together" return True, "" def _apply_docopt(self, argv: list, is_json: bool) -> Tuple[dict, str]: """ Applies docopt to parse utility arguments. Performs checks before and after parsing """ ok, message = self._perform_checks_before_docopt(argv) if not ok: return {}, message try: args = docopt(self.docstring, argv) except DocoptExit: s_error_string = "ERROR: Invalid parameter passed or required parameter is missing" if not is_json: s_error_string += "\n\n" + self.docstring return {}, s_error_string ok, message = self._perform_checks_after_docopt(args) if not ok: return {}, message return args, "" @staticmethod def _get_schema() -> Schema: """ Returns schema that should be used to validate parsed arguments """ return global_schema @staticmethod def _validate_schema(schema: Schema, args: dict) -> Tuple[dict, str]: """ Validate parsed arguments with provided schema """ try: args = schema.validate(args) except SchemaError as e: return {}, str(e) return args, "" def parse_args(self, argv: list, is_json: bool = False) -> Tuple[bool, Union[str, dict]]: """ Parses and validates arguments passed to the utility """ args, message = self._apply_docopt(argv, is_json) if not args: return False, message # if command was not specified, use `get` command if not any(args.get(command) for command in self.commands): args["get"] = True schema = self._get_schema() args, message = self._validate_schema(schema, args) if not args: return False, message return True, args class CloudlinuxLimitsNoLveOptsParser(CloudlinuxLimitsOptsParser): docstring = f"""Utility to get/set any Cloudlinux limits Usage: {PROG_NAME} set [--json] (--lve-id <int> \| --username <str>) (--cagefs <enabled,disabled>) {PROG_NAME} set [--json] (--lve-id <int> \| --username <str>) (--unlimited) {PROG_NAME} set [--json] (--lve-id <int> \| --username <str>) (--default <str>) {PROG_NAME} set [--json] (--lve-id <int> \| --username <str>) [--inodes <N,M> --save-all-parameters] {PROG_NAME} [get] [--json] [--lve-id <int> \| --username <str> \| --domain <str>] [--limits=<keys>] {PROG_NAME} (-h \| --help) Options: --json Return data in JSON format. --lve-id <int> LVE id. will display record only for that LVE id. --username <str> Execute command only for specific user. --cagefs <enabled\|disabled> Enable or disable CageFS for a user. --unlimited Set supported limits to unlimited. --default [limits] Reset supported limits to the Package defaults. List of comma-separated limits to reset them to default or "all" --domain <str> Show data only for specific domain --limits <keys> Available keys: inodes --inodes <N,M> Set inode limits. N - soft, M - hard. --save-all-parameters Save all parameters even if they match with defaults settings. -h, --help Show this help message and exit """ commands = ("set", "get") def _perform_checks_before_docopt(self, argv: list) -> Tuple[bool, str]: # no checks to be performed return True, "" def _perform_checks_after_docopt(self, args: dict) -> Tuple[bool, str]: # no checks to be performed return True, "" @staticmethod def _get_schema() -> Schema: # limit schema to the given options only, other options are not needed schema_options = {"set", "get", "--json", "--lve-id", "--username", "--cagefs", "--domain", "--unlimited", "--default", "--limits", "--inodes", "--save-all-parameters", "--help"} options = { option: value for option, value in global_schema.schema.items() if option in schema_options } return Schema(options) def parse_args(self, argv: list, is_json: bool = False) -> Tuple[bool, Union[str, dict]]: ok, args = super().parse_args(argv, is_json) if not ok: return False, args # fill parsed arguments dictionary with params from global schema # using negative values (False or None) # do this because these parametes can be checked in the utility for key, value in global_schema.schema.items(): args.setdefault(key, False if value is bool else None) return True, args	15.99 KB	2026-02-17 10:40
📄 limits.py	50.63 KB	2026-02-17 10:40
📄 utils.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 json import os import sys import time from typing import Optional from clcommon.utils import run_command, ExternalProgramFailed from clcommon.clfunc import memory_to_page, page_to_memory from clcommon.clexception import FormattedException VALUES_STR = 'Available values for option' _LIMITS_LOWER_BOUNDS_CONFIG_FILE = '/etc/sysconfig/limits_lower_bounds' class LowerBoundsConfigException(FormattedException): pass def _read_lower_bounds_config() -> dict: """ Read all lower bounds from config file Returns dict with lower bound names as keys and values as strings :raises LowerBoundsConfigException: If file exists but cannot be read """ # If file doesn't exist return empty dict if not os.path.exists(_LIMITS_LOWER_BOUNDS_CONFIG_FILE): return {} lower_bounds = {} try: with open(_LIMITS_LOWER_BOUNDS_CONFIG_FILE, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if line and '=' in line and not line.startswith('#'): key, value = line.split('=', 1) lower_bounds[key.strip()] = value.strip() except (OSError, IOError) as e: raise LowerBoundsConfigException({ 'message': "Failed to read limits' lower bounds configuration from %(file)s. " "The file may be corrupted. Error: %(error)s", 'context': { 'file': _LIMITS_LOWER_BOUNDS_CONFIG_FILE, 'error': str(e) } }) from e return lower_bounds def _write_lower_bounds_config(lower_bounds: dict) -> bool: """ Write lower bounds to config file """ try: os.makedirs(os.path.dirname(_LIMITS_LOWER_BOUNDS_CONFIG_FILE), exist_ok=True) with open(_LIMITS_LOWER_BOUNDS_CONFIG_FILE, 'w', encoding='utf-8') as f: f.write('# CloudLinux lower bounds configuration\n') f.write('# Format: RESOURCE_TYPE=value\n\n') for key, value in sorted(lower_bounds.items()): f.write(f'{key}={value}\n') return True except (OSError, IOError): return False def read_lower_bound(resource_type: str) -> Optional[int]: """ Read lower bound for any resource type from config file Returns lower bound in pages or None if not set :param resource_type: Type of resource (e.g., 'PMEM', 'VMEM') :raises LowerBoundsConfigException: If config file is corrupted """ lower_bounds = _read_lower_bounds_config() limit_str = lower_bounds.get(resource_type.upper()) if limit_str: return memory_to_page(limit_str) return None def write_lower_bound(resource_type: str, bound_value: str) -> bool: """ Write lower bound for any resource type to config file :param resource_type: Type of resource (e.g., 'PMEM', 'VMEM') :param bound_value: Lower bound value as string (e.g., '512M', '1G') :raises LowerBoundsConfigException: If config file is corrupted """ lower_bounds = _read_lower_bounds_config() lower_bounds[resource_type.upper()] = bound_value return _write_lower_bounds_config(lower_bounds) def check_pmem_lower_bound(pmem_pages: int) -> Optional[dict]: """ Check if PMEM value is below configured lower bound and return error :param pmem_pages: PMEM value in pages :return: Error dict with 'message' and 'context' or None """ try: lower_bound_pages = read_lower_bound('PMEM') except LowerBoundsConfigException as e: return {'message': e.message, 'context': e.context} if lower_bound_pages is not None and pmem_pages != 0 and pmem_pages < lower_bound_pages: lower_bound = page_to_memory(lower_bound_pages) return { 'message': ( 'PMEM limit is below configured lower bound of %(bound)s. ' 'This may cause application instability and performance issues.' ), 'context': {'bound': lower_bound} } return None def check_pmem_lower_bound_from_string(pmem_value: str) -> Optional[dict]: """ Check if PMEM value (as string) is below configured lower bound and return error :param pmem_value: PMEM value as string (e.g., "5m", "512M", etc.) :return: Error dict with 'message' and 'context' or None """ pmem_pages = memory_to_page(pmem_value) if pmem_pages is not None: return check_pmem_lower_bound(pmem_pages) return None def replace_params(data): """ Replacing params in data for show error message :param data: error's data for show message :return: """ if data.startswith("--"): param, text = data.split(" ", 1) return {"result": "%(param)s " + text, "context": {"param": param}} if data.startswith(VALUES_STR): text, param = data.split(":", 1) return {"result": text + ": %(available_options)s", "context": {"available_options": param.strip()}} return {"result": data} def _is_string_number(s_val): """ Checks is string contains a number (integer or float) :param s_val: String to check :return: True - string is number, False - not number """ try: float(s_val) return True except ValueError: return False def convert_mem_value_to_bytes(value): """ Convert value in Gbytes,Mbytes to bytes :param value: value of mem limit :return: value in bytes """ value = str(value).lower() if value.endswith('k'): power = 1 elif value.endswith('m'): power = 2 elif value.endswith('g'): power = 3 elif _is_string_number(value): power = 1 value = f'{value}k' else: raise ValueError('Wrong memory value') return int(1024 ** power * float(value[:-1])) def _convert_memory_value_to_adaptive_format(value, convert=True): """ Convert memory value to adaptive value in GB, TB, etc :param value: memory value in MB or KB :param convert: if True - convert value, False - not convert :return: adaptive value in GB, TB, etc """ if not convert: return value value = str(value).lower() units = ['K', 'M', 'G', 'T', 'P'] if value.endswith('m'): del units[0] value = str(value).lower().replace('m', '').replace('k', '') # remove unit symbol if value.startswith(''): result = '' value = value.replace('*', '') else: result = '' value = float(value) for unit in units: if value // 1024 > 0: value /= 1024 elif value == 0: result = f'{result}0K' break else: result = f'{result}{value:.2f}{unit}' break return result def print_dictionary(data_dict, is_json=False): """ Print specified dictionary :param data_dict: data dictionary to print :param is_json: True - print in JSON, False - in text :return: None """ # Print as JSON # print json.dumps(data_dict, indent=2, sort_keys=True) if is_json: # Print as JSON print(json.dumps(data_dict, sort_keys=True)) else: # Print as text print(data_dict) def print_error_and_exit(message, is_json=False): """ Prints to stdout :param: is_json - True if print error in json format, False - text """ data = {"timestamp": time.time(), "result": str(message)} print_dictionary(data, is_json) sys.exit(1) def is_quota_supported(cl_quota_path, repquota_path): """ Detect quota is supported :return: True/False - quotas supported/not supported """ # Check if all needed utilities present if not os.path.isfile(cl_quota_path) or not os.path.isfile(repquota_path): return False return True def is_quota_active(cl_quota_path, repquota_path): """ Detect quota is activated :return: True/False - quotas activated/not activated """ # If quotas not supported they are not activated if not is_quota_supported(cl_quota_path, repquota_path): return False # Check is quota activated cmd = [repquota_path, '-nva'] try: stdout = run_command(cmd) except ExternalProgramFailed: return False # quotas not supported if repqouta returns nothing if not stdout: return False return True	8.39 KB	2026-02-17 10:40

Terminal