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 .limits_validator import LimitsValidator from .base import ENTITY_USER, ENTITY_PACKAGE, ENTITY_RESELLER, ENTITY_DEFAULTS, ENTITY_BUILTIN __all__ = ('LimitsValidator', 'ENTITY_USER', 'ENTITY_PACKAGE', 'ENTITY_RESELLER', 'ENTITY_DEFAULTS', 'ENTITY_BUILTIN')	460 B	2026-02-17 10:40
📁 __pycache__	-	2026-03-27 02:38
📄 base.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 typing import Any, Optional, Dict # NOQA from cllimitslib_v2 import LimitsDataStorage ENTITY_USER = 'user' ENTITY_PACKAGE = 'package' ENTITY_RESELLER = 'reseller' ENTITY_DEFAULTS = 'defaults' ENTITY_BUILTIN = 'builtin' class BaseValidator: def __init__(self, _limits_data_storage=None): # type: (LimitsDataStorage) -> None self.limits_provider = _limits_data_storage or LimitsDataStorage() self._message_dict = { 'common_msg': None, 'specific_msg': None, 'affected_entity': None, 'entity_id': None, } def validate_existing_limits(self, args, kwargs): # type: (Any, *Any) -> Optional[bool, Dict] """ Validate limits which already are recorded in ve.cfg """ raise NotImplementedError() def validate_input_limits(self, args, *kwargs): # type: (Any, *Any) -> Optional[bool, Dict] """ Validate limits which we want to set """ raise NotImplementedError() def _validate(self, args, *kwargs): # type: (Any, **Any) -> Optional[bool, Dict] raise NotImplementedError()	1.34 KB	2026-02-17 10:40
📄 ep_nproc_validator.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 typing import Dict, Optional, Tuple, Union # NOQA from cllimitslib_v2 import LimitsDataStorage, DEFAULTS # NOQA from cllimitslib_v2.lve_storage import ( Package, User, Reseller, OWN_SOURCE, RESELLER_SOURCE, # NOQA PACKAGE_SOURCE, GLOBAL_SOURCE ) from .base import BaseValidator, ENTITY_USER, ENTITY_PACKAGE, ENTITY_RESELLER, ENTITY_DEFAULTS class EpNprocValidator(BaseValidator): def __init__(self, _limits_data_storage=None): # type: (LimitsDataStorage) -> None super().__init__(_limits_data_storage=_limits_data_storage) # difference beetween nproc limit and ep for normal server work # see LU-919 self.diff_nproc_and_ep = 15 @property def message_dict(self): # type: () -> Dict common_msg = ( f'NPROC limit must be greater than EP + {self.diff_nproc_and_ep} limit, because ' 'number of processes and threads within LVE includes ' 'also Apache processes/threads, SSH sessions and etc, ' 'which enter into LVE.' ) message_dict = dict(self._message_dict) message_dict['common_msg'] = common_msg return message_dict def _ep_and_nproc_is_valid(self, ep, nproc): # type: (int, int) -> bool if nproc == 0 or ep == 0: result = True else: result = nproc >= ep + self.diff_nproc_and_ep return result def _validate_nproc_and_ep_for_user(self, user, ep, nproc, has_input_limits): # type: (User, Union[int, None], Union[int, None], bool) -> Tuple[bool, Dict] """ :param has_input_limits: User has input limits, which will be set to it """ message_dict = self.message_dict user_limits = self.limits_provider.get_user_limits(user) ep = int(ep or user_limits.ep.value) nproc = int(nproc or user_limits.nproc.value) result = self._ep_and_nproc_is_valid(ep, nproc) ep_source = user_limits.ep.source nproc_source = user_limits.nproc.source if has_input_limits: message_dict['affected_entity'] = ENTITY_USER message_dict['entity_id'] = user.id elif OWN_SOURCE in (nproc_source, ep_source): message_dict['affected_entity'] = ENTITY_USER message_dict['entity_id'] = user.id if ep_source != OWN_SOURCE: inheritance_limit = 'ep' elif nproc_source != OWN_SOURCE: inheritance_limit = 'nproc' else: inheritance_limit = None message_dict['specific_msg'] = None if inheritance_limit is None else ( f'{inheritance_limit.upper()} is inherited ' f'from {getattr(user_limits, inheritance_limit).source.capitalize()}' ) elif OWN_SOURCE not in (nproc_source, ep_source) and nproc_source == ep_source: message_dict['affected_entity'] = ep_source if ep_source == PACKAGE_SOURCE: message_dict['entity_id'] = user.package.name elif ep_source == RESELLER_SOURCE: message_dict['entity_id'] = user.reseller.name elif ep_source == GLOBAL_SOURCE: message_dict['entity_id'] = ENTITY_DEFAULTS elif OWN_SOURCE not in (nproc_source, ep_source) and nproc_source != ep_source: message_dict['affected_entity'] = ENTITY_USER message_dict['entity_id'] = user.id message_dict['specific_msg'] = ( 'Both validated limits (EP, NPROC) inheritance ' f'from different sources: {ep_source} and {nproc_source}' ) return result, message_dict def _validate_nproc_and_ep_for_package(self, package, ep, nproc, has_input_limits): # type: (Package, Union[int, None], Union[int, None], bool) -> Tuple[bool, Dict] """ :param has_input_limits: User has input limits, which will be set to it """ message_dict = self.message_dict package_limits = self.limits_provider.get_package_limits(package) ep = int(ep or package_limits.ep.value) nproc = int(nproc or package_limits.nproc.value) result = self._ep_and_nproc_is_valid(ep, nproc) ep_source = package_limits.ep.source nproc_source = package_limits.nproc.source if has_input_limits: message_dict['affected_entity'] = ENTITY_PACKAGE message_dict['entity_id'] = package.name elif OWN_SOURCE in (nproc_source, ep_source): message_dict['affected_entity'] = ENTITY_PACKAGE message_dict['entity_id'] = package.name if ep_source != OWN_SOURCE: inheritance_limit = 'ep' elif nproc_source != OWN_SOURCE: inheritance_limit = 'nproc' else: inheritance_limit = None message_dict['specific_msg'] = None if inheritance_limit is None else ( f'{inheritance_limit.upper()} is inherited ' f'from {getattr(package_limits, inheritance_limit).source.capitalize()}' ) elif OWN_SOURCE not in (nproc_source, ep_source) and nproc_source == ep_source: message_dict['affected_entity'] = ep_source if ep_source == RESELLER_SOURCE: message_dict['entity_id'] = package.provider.name elif ep_source == GLOBAL_SOURCE: message_dict['entity_id'] = ENTITY_DEFAULTS elif OWN_SOURCE not in (nproc_source, ep_source) and nproc_source != ep_source: message_dict['affected_entity'] = ENTITY_PACKAGE message_dict['entity_id'] = package.name message_dict['specific_msg'] = ( 'Both validated limits (EP, NPROC) inheritance ' f'from different sources: {ep_source} and {nproc_source}' ) return result, message_dict def _validate_nproc_and_ep_for_reseller(self, reseller, ep, nproc, has_input_limits): # type: (Reseller, Union[int, None], Union[int, None], bool) -> Tuple[bool, Dict] """ :param has_input_limits: User has input limits, which will be set to it """ message_dict = self.message_dict reseller_limits = self.limits_provider.get_reseller_limits(reseller) ep = int(ep or reseller_limits.ep.value) nproc = int(nproc or reseller_limits.nproc.value) result = self._ep_and_nproc_is_valid(ep, nproc) ep_source = reseller_limits.ep.source nproc_source = reseller_limits.nproc.source if has_input_limits: message_dict['affected_entity'] = ENTITY_RESELLER message_dict['entity_id'] = reseller.name elif OWN_SOURCE in (nproc_source, ep_source): message_dict['affected_entity'] = ENTITY_RESELLER message_dict['entity_id'] = reseller.name if ep_source != OWN_SOURCE: inheritance_limit = 'ep' elif nproc_source != OWN_SOURCE: inheritance_limit = 'nproc' else: inheritance_limit = None message_dict['specific_msg'] = None if inheritance_limit is None else ( f'{inheritance_limit.upper()} is inherited ' f'from {getattr(reseller_limits, inheritance_limit).source.capitalize()}' ) return result, message_dict def _validate_nproc_and_ep_for_defaults(self, reseller, ep, nproc, has_input_limits): # type: (Union[Reseller, DEFAULTS], Union[int, None], Union[int, None], bool) -> Tuple[bool, Dict] """ :param has_input_limits: User has input limits, which will be set to it """ message_dict = self.message_dict id_entity = reseller.name if isinstance(reseller, Reseller) else DEFAULTS default_limits = self.limits_provider.get_defaults_limits(reseller) ep = int(ep or default_limits.ep.value) nproc = int(nproc or default_limits.nproc.value) result = self._ep_and_nproc_is_valid(ep, nproc) ep_source = default_limits.ep.source nproc_source = default_limits.nproc.source if has_input_limits: message_dict['affected_entity'] = ENTITY_DEFAULTS message_dict['entity_id'] = id_entity elif OWN_SOURCE in (nproc_source, ep_source): message_dict['affected_entity'] = ENTITY_DEFAULTS message_dict['entity_id'] = id_entity if ep_source != OWN_SOURCE: inheritance_limit = 'ep' elif nproc_source != OWN_SOURCE: inheritance_limit = 'nproc' else: inheritance_limit = None message_dict['specific_msg'] = None if inheritance_limit is None else ( f'{inheritance_limit.upper()} is inherited ' f'from {getattr(default_limits, inheritance_limit).source.capitalize()}' ) return result, message_dict def _validate(self, validated_entity, ep, nproc, has_input_limits, check_reseller_defaults=False): # type: (Union[User, Reseller, Package, None], Optional[int], Optional[int], bool, bool) -> Tuple[bool, Dict] message_dict = self.message_dict result = True if isinstance(validated_entity, User): result, message_dict = self._validate_nproc_and_ep_for_user( validated_entity, ep, nproc, has_input_limits ) elif isinstance(validated_entity, Package): result, message_dict = self._validate_nproc_and_ep_for_package( validated_entity, ep, nproc, has_input_limits ) elif isinstance(validated_entity, Reseller) and not check_reseller_defaults: # we validate reseller limits result, message_dict = self._validate_nproc_and_ep_for_reseller( validated_entity, ep, nproc, has_input_limits ) elif isinstance(validated_entity, Reseller) and check_reseller_defaults or validated_entity == DEFAULTS: # we validate default limits (include reseller defaults) result, message_dict = self._validate_nproc_and_ep_for_defaults( validated_entity, ep, nproc, has_input_limits ) return result, message_dict def validate_existing_limits(self, validated_entity, check_reseller_defaults=False): # type: (Union[User, Reseller, Package, None], bool) -> Tuple[bool, Dict] """ Validate limits which already are recorded in ve.cfg """ result, message_dict = self._validate(validated_entity, None, None, False, check_reseller_defaults) return result, message_dict def validate_input_limits(self, validated_entity, input_limits, check_reseller_defaults=False): # type: (Union[User, Reseller, Package, None], Dict, bool) -> Tuple[bool, Dict] """ Validate limits which we want to set """ message_dict = self.message_dict nproc = input_limits.get('nproc', None) ep = input_limits.get('ep', None) # if input_limits has no ep and nproc we set has_settable_limits to None # It means nothing to validate has_input_limits = nproc is not None or ep is not None # We skip check if nproc is unlimited or input limits don't have ep and nproc values if nproc == 0 or not has_input_limits: result = True else: result, message_dict = self._validate(validated_entity, ep, nproc, has_input_limits, check_reseller_defaults) return result, message_dict	11.86 KB	2026-02-17 10:40
📄 limits_validator.py # coding: utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from collections import defaultdict from itertools import chain from typing import Any, Dict, List, Optional, Tuple, Union # NOQA from cllimits import LveCtl from cllimitslib_v2 import DEFAULTS, LimitsDataStorage from clveconfig.ve_config_reader import DEFAULT_PROVIDER from .base import ENTITY_DEFAULTS, ENTITY_PACKAGE, ENTITY_RESELLER, ENTITY_USER, BaseValidator # NOQA from .ep_nproc_validator import EpNprocValidator class LimitsValidator: # This dict contains keys which are old names of entities # and contais values which are new names of entities. # Old names will be replaced by new names for improve readability of error message dict_of_replaced_names_of_entities = { 'user': 'User(s)', 'package': 'Package(s)', 'reseller': 'Reseller(s)', 'defaults': 'Defaults of resellers' } # This dict contains keys which are old ids of entities # and contais values which are new ids of entities. # Old ids will be replaced by ids names for improve readability of error message dict_of_replaced_ids_of_entities = { 'defaults': 'Hoster defaults limits' } # This list contains ids entities which will be deleted # for improve readability of error message list_of_deleted_ids_of_entities = [ DEFAULTS ] # This dict contais keys which are old specific messages and # contains values which are new specific message # Old specific msgs will be replaced by new for improve readability of error message dict_of_replaced_specific_messages = { 'inherited from Global': 'inherited from Hoster' } def __init__(self, _limits_data_storage=None, validators=None): # type: (Optional[LimitsDataStorage], Optional[List[BaseValidator]]) -> None self.limits_provider = _limits_data_storage or LimitsDataStorage() self._list_validators = validators or [ EpNprocValidator(self.limits_provider), ] # type: List[BaseValidator] self.message_dict = { 'common_msg': None, 'specific_msg': None, 'affected_entity': None, 'entity_id': None, } @staticmethod def _format_message_string_from_single_msg_dict(msg_dict): # type: (Dict[str, str, str, str]) -> str """ Format message string from single message dict. That function should be used for formatting message while validation input limits """ common_msg = msg_dict['common_msg'] specific_msg = msg_dict['specific_msg'] affected_entity = msg_dict['affected_entity'] entity_id = msg_dict['entity_id'] if affected_entity is not None and entity_id is not None: if entity_id == DEFAULTS: msg_about_affect = 'Hoster defaults limits is affected' elif affected_entity == ENTITY_DEFAULTS: msg_about_affect = f'Defaults limits of Reseller "{entity_id}" are affected.' elif affected_entity == ENTITY_USER: msg_about_affect = f'Limits of user with LVE ID "{entity_id}" are affected.' elif affected_entity == ENTITY_RESELLER: msg_about_affect = f'Limits of Reseller "{entity_id}" are affected.' else: msg_about_affect = f'Limits of package "{entity_id}" are affected.' else: msg_about_affect = '' if specific_msg is None: specific_msg = '' result_msg = f'{common_msg} {msg_about_affect} {specific_msg}' return result_msg def _call_validators_and_process_result(self, only_input_limits, args, kwargs): # type: (bool, Any, *Any) -> Tuple[bool, List] """ Call each validator from validators list and processing results args and kwargs are argumets for calling of validator We validate limits which already are recorded in ve.cfg if only_input_limits is False. We validate limits which we want to set if only_input_limits is True We return aggregated result as bool value and list of messages which explain which limits are wrong and who holds those """ msg_list = [] total_result = True for validator in self._list_validators: if only_input_limits: result, msg_dict = validator.validate_input_limits(args, *kwargs) else: result, msg_dict = validator.validate_existing_limits(args, *kwargs) total_result &= result msg_list.append(msg_dict) return total_result, msg_list def validate_input_limits(self, entity_type, entity_id, input_limits, reseller=DEFAULT_PROVIDER): # type: (str, Union[str, int], Dict, str) -> Tuple[bool, List] check_reseller_defaults = False if entity_type == ENTITY_USER: validated_entity = self.limits_provider.get_user_by_uid(int(entity_id)) elif entity_type == ENTITY_PACKAGE: validated_entity = self.limits_provider.get_package_by_name_and_reseller(str(entity_id), str(reseller)) elif entity_type == ENTITY_RESELLER: validated_entity = self.limits_provider.get_reseller_by_name(str(entity_id)) elif entity_type == ENTITY_DEFAULTS: if entity_id != DEFAULTS: validated_entity = self.limits_provider.get_reseller_by_name(str(entity_id)) else: validated_entity = entity_id check_reseller_defaults = True else: validated_entity = None total_result, msg_list = self._call_validators_and_process_result( True, validated_entity, input_limits, check_reseller_defaults, ) msg_list = [self._format_message_string_from_single_msg_dict(msg_dict) for msg_dict in msg_list] return total_result, msg_list def _validate_users(self): # type: () -> List[Tuple[bool, List[Dict]]] users = self.limits_provider.users.values() result = [self._call_validators_and_process_result(False, user) for user in users] return result def _validate_packages(self): # type: () -> List[Tuple[bool, List[Dict]]] packages = self.limits_provider.packages.values() result = [self._call_validators_and_process_result(False, package) for package in packages] return result def _validate_resellers(self): # type: () -> List[Tuple[bool, List[Dict]]] resellers = self.limits_provider.resellers.values() # We should validate only activated resellers, # because not activated resellers don't have any limits. result = [self._call_validators_and_process_result(False, reseller) for reseller in resellers if reseller.defaults is not None] return result def _validate_defaults(self): # type: () -> List[Tuple[bool, List[Dict]]] result = [] for default_name in self.limits_provider.defaults.keys(): default_entity = self.limits_provider.resellers.get(default_name, DEFAULTS) # We should validate only defaults which belong to activated resellers, # because not activated resellers don't have any limits. if default_entity != DEFAULTS and default_entity.defaults is None: continue result.append(self._call_validators_and_process_result(False, default_entity)) return result def _replace_specific_msgs(self, specific_msg): # type: (str) -> str """ Replacing some specific messages for improve readability of error message """ for old_spec_msg, new_spec_msg in self.dict_of_replaced_specific_messages.items(): if specific_msg is None: break specific_msg = specific_msg.replace(old_spec_msg, new_spec_msg) return specific_msg def _replace_ids_of_entities(self, entity_id): # type: (str) -> str """ Replacing some ids of entities for improve readability of error message """ if entity_id in self.dict_of_replaced_ids_of_entities: return self.dict_of_replaced_ids_of_entities[entity_id] return entity_id def _replace_names_of_entities(self, entity_name): # type: (str) -> str """ Replacing some names of entities for improve readability of error message """ if entity_name in self.dict_of_replaced_names_of_entities: return self.dict_of_replaced_names_of_entities[entity_name] return entity_name def _format_message_string_for_existing_limits(self, total_result_list): # type: (List[Dict[str, str, str, str]]) -> str """ Format message string from result of execution of validation the existing limits """ total_result_dict = defaultdict(lambda: defaultdict(list)) total_msg = None # We are groupping affected entities (entity_id) by name of check (common_msg) # and type of affected entity (affected_entity). for msg_dict in total_result_list: name_of_check = msg_dict['common_msg'] type_of_affected_entity = self._replace_names_of_entities(msg_dict['affected_entity']) entity_id = self._replace_ids_of_entities(msg_dict['entity_id']) # Deleting some ids of entities for improve readability of error message if entity_id in self.list_of_deleted_ids_of_entities: continue total_result_dict[name_of_check][type_of_affected_entity].append({ 'entity_id': entity_id, 'specific_msg': self._replace_specific_msgs(msg_dict['specific_msg']) }) for common_msg, entities_dict in total_result_dict.items(): total_msg = f'{"" if total_msg is None else total_msg}{common_msg}\n' for entity_type, list_entity_dicts in entities_dict.items(): affected_entities = [] # make string as `entity_1 (specific_msg_1), entity_2, entity_3, entity_4 (specific_msg) for entity_dict in list_entity_dicts: if entity_dict['specific_msg'] is None: affected_entity = entity_dict['entity_id'] else: affected_entity = f'{entity_dict["entity_id"]} ({entity_dict["specific_msg"]})' affected_entities.append(affected_entity) affected_entities = ', '.join(str(item) for item in set(affected_entities)) total_msg = f'{total_msg}{entity_type.capitalize()}: {affected_entities}\n' return total_msg def validate_existing_limits(self): # type: () -> Optional[str] total_result_list = [ msg_dict for result, msg_dict_list in chain( self._validate_users(), self._validate_packages(), self._validate_resellers(), self._validate_defaults(), ) for msg_dict in msg_dict_list if not result ] total_msg = self._format_message_string_for_existing_limits(total_result_list) return total_msg @staticmethod def is_low_pmem_limit_present() -> bool: """ Check is Low PMEM limit present :return True/False """ _lvectl = LveCtl() _panel_uids_list = _lvectl.get_panel_users_uid_list() _lvectl._load_info(False) for uid in _panel_uids_list: if uid == 0: # skip defaults continue limits = _lvectl.get_limits_by_user_id(uid) # limits example: # {'cpu': {'all': '100'}, 'io': {'all': '1024'}, 'vmem': '0', 'ep': '20', 'pmem': '268435456', # 'nproc': '100', 'iops': '4096'} try: # Extract PMEM limit pmem_limit_bytes = int(limits['pmem'].replace('', '')) if pmem_limit_bytes != 0 and pmem_limit_bytes < 512 * 1024 * 1024: # PMEM limit not unlimited and < 512 MB - error return True except ValueError: pass return False	12.21 KB	2026-02-17 10:40

Terminal