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 #!/opt/cloudlinux/venv/bin/python3 -bb # -- 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 configparser as ConfigParser import csv import fcntl import os import pwd import re import tempfile from collections import defaultdict from stat import S_IRGRP, S_IROTH, S_IRUSR, S_IWUSR, ST_DEV from typing import Dict, List, Optional, Tuple import clcontrollib import cldetectlib from clcommon import FormattedException # pylint: enable=E0611 from clcommon.clpwd import ClPwd from clcommon.clquota import check_quota_enabled from clcommon.cpapi import admin_packages, list_users, resellers_packages from clcommon.cpapi.cpapiexceptions import CPAPIExternalProgramFailed, EncodingError from clcommon.utils import ( ExternalProgramFailed, get_file_lines, get_filesystem_type, run_command, write_file_lines, ) IS_DA = clcontrollib.detect.is_da() DEFAULT_PACKAGE = 'default' VE_DEFAULT_PACKAGE = 'VE_DEFAULT' # virtual package, alias for uid=0 class NoSuchPackageException(Exception): def __init__(self, package): Exception.__init__(self, "No such package (%s)" % (package,)) class NoSuchUserException(Exception): def __init__(self, user): Exception.__init__(self, "No such user (%s)" % (user,)) class InsufficientPrivilegesException(Exception): def __init__(self): Exception.__init__(self, "Insufficient privileges") class IncorrectLimitFormatException(Exception): def __init__(self, limit): Exception.__init__(self, "Incorrect limit format (%s)" %(limit,)) class MalformedConfigException(FormattedException): """ Raised when config files is malformed and cl-quota is not able to work with it """ def __init__(self, error: ConfigParser.ParsingError): super(MalformedConfigException, self).__init__({ 'message': "cl-quota can't work because for malformed config. " "Please, contact CloudLinux support if you " "need help with resolving this issue. " "Details: %(error_message)s", 'context': dict( error_message=str(error) ) }) class GeneralException(Exception): def __init__(self, message): Exception.__init__(self, message) class QuotaDisabledException(Exception): def __init__(self): super(QuotaDisabledException, self).__init__('Quota disabled for all users on server') class UserQuotaDisabledException(QuotaDisabledException): """ Raised when quota is disabled for one particular user """ def __init__(self, uid=None, homedir=None, message=None): all_msg = 'Quota disabled' if uid: all_msg += ' for user id %s' % uid if homedir: all_msg += ' (home directory %s)' % homedir if message: all_msg += '; %s' % message Exception.__init__(self, all_msg) def _is_sys_path(path): """ >>> _is_sys_path('/home/username') False >>> _is_sys_path('/var/davecot') True """ if path[-1] != '/': path += '/' sys_path_ = ('/root/', '/usr/', '/var/', '/sbin/', '/dev/', '/bin/', '/srv/', '/sys/', '/etc/ntp/') if path == '/': return True for path_ in sys_path_: if path.startswith(path_): return True def _get_users_list(): """ Return no system users uid list """ cl_pwd = ClPwd() pw_dict = cl_pwd.get_user_dict() users_uid = [pw_dict[usr].pw_uid for usr in pw_dict if not _is_sys_path(pw_dict[usr].pw_dir)] return users_uid def is_quota_inheritance_enabled() -> bool: """ Check `cl_quota_inodes_inheritance` parameter in the config file """ res = cldetectlib.get_boolean_param(cldetectlib.CL_CONFIG_FILE, 'cl_quota_inodes_inheritance', default_val=False) return res class QuotaWrapper(object): """ Base quota class for inode quotas handling * Update system quotas via setquota * Retrieves system quotas via repquota * Stores current quotas in /etc/container/cl-quotas.dat file * Maintaines /etc/container/cl-quotas.cache file with resolved quotas. That file can be read by non-privileged users """ PROC_MOUNTS = '/proc/mounts' QUOTASYNC = '/usr/bin/quotasync' SETQUOTA = '/usr/sbin/setquota' REPQUOTA = '/usr/sbin/repquota' GETPACKS = '/usr/bin/getcontrolpaneluserspackages' DATAFILE = '/etc/container/cl-quotas.dat' CACHEFILE = '/etc/container/cl-quotas.cache' # File lock variables LOCK_FD = None LOCK_FILE = DATAFILE + '.lock' LOCK_WRITE = False def __init__(self): self._assert_file_exists(QuotaWrapper.PROC_MOUNTS) self._assert_file_exists(QuotaWrapper.REPQUOTA) self._assert_file_exists(QuotaWrapper.SETQUOTA) self._quota_enabled_list = list() self._panel_present = None self._grace = {} self._quota = {} self._device_quota = {} self._package_to_uids_map = {} # type: Dict[str, List[str]] self._uid_to_packages_map = {} # type: Dict[str, List[str]] self._uid_to_homedir_map = {} # type: Dict[str, str] self._dh = self._get_saved_data_handler() self._fields = ['bytes_used', 'bytes_soft', 'bytes_hard', 'inodes_used', 'inodes_soft', 'inodes_hard'] self._euid = os.geteuid() self._devices = self._load_quota_devices() self._mountpoint_device_mapped = self._get_mountpoint_device_map(self._devices) self._device_user_map = None # List of all packages (all admin's packages + all reseller packages) self._all_package_list = None @staticmethod def _assert_file_exists(path): """ Checks if command is present and exits if no """ if not os.path.exists(path): raise RuntimeError('No such command (%s)' % (path,)) def __enter__(self): return self def __exit__(self, _type, _value, _traceback): if self.LOCK_FD is not None: self.LOCK_FD.close() def get_user_limits(self, uid): ''' Returns user limits converted to tuples ''' return self._convert_data_to_tuples(self._get_current_quotas(uid)) def get_all_users_limits(self): ''' Returns all user limits converted to tuples ''' return self._convert_data_to_tuples(self._get_current_quotas()) def get_package_limits(self, package): """ :param packname: Package name for get limits. If None, returns all packages, else - only supplied package Returns package limits converted to tuples (called only from main) """ return self._convert_data_to_tuples(self._get_package_quotas(packname=package)) def get_all_packages_limits(self, package=None): """ Returns all packages limits converted to tuples (called only from main) """ return self._convert_data_to_tuples(self._get_package_quotas(packname=package, all_packages=True)) def _preprocess_limit(self, limit): """ Preprocessed passed limit: 'default' --> '0', 'unlimited' --> -1, else calls _check_limit :param limit: :return: """ if limit == 'default': return '0' if limit in ('unlimited', '-1'): return '-1' return self._check_limit(limit) def _get_package_from_dh(self, package): return self._dh.get('packages', package).split(':') def _get_all_packages_with_limits(self, clean_dead_packages=False): """ Retrive all available packages with their limits :param clean_dead_packages: if True - remove all nonexistent packages from cl-quotas.dat :return: Dictionary: { 'package_name': (soft_limit, hard_limit) } """ # result dictionary package_limits_dict = {} # Load packages limits from cl-quota.dat db_packages = {} if self._dh.has_section('packages') and len(self._dh.items('packages')) > 0: list_of_packages = self._get_all_package_list() for package in self._dh.options('packages'): if clean_dead_packages and package not in list_of_packages: self._dh.remove_option('packages', package) continue package_limits = self._get_package_from_dh(package) # Pass package, if limits not well-formed if len(package_limits) != 2: continue db_packages[package] = package_limits[0], package_limits[1] if clean_dead_packages: self._write_data() # Put all panel packages to result dictionary self._get_package_to_users_map() for package in self._package_to_uids_map.keys(): if package in db_packages: # if package present in cl-quota.dat, take limits package_limits_dict[package] = db_packages[package] else: package_limits_dict[package] = ('0', '0') return package_limits_dict def set_user_limit( self, uid: str, soft: Optional[str] = None, hard: Optional[str] = None, save: bool = True, force_save: bool = False, only_store: bool = False, ): """ Set limits for users * Resolve limits according to those saved in cl-quota.dat Limits are resolved in the following order: user limits --> package limits --> root user (uid=0) limits * Apply new limits when resolved limits differ from those in cache or when `force_save` is True * Write updated values to cl-quota.dat if `save` is True Always update cl-quota.dat in case both zeroes or both unlimited are provided :param uid: user id :param soft: soft limit value :param hard: hard limit value :param save: save limits to cl-quota.dat :param force_save: save limits to cl-quota.dat even if they are not changed :param only_store: store limits in memory, but do not apply them :return: None """ self._check_admin() # Validate limits value, convert 'unlimited' --> '-1', 'default' --> '0' soft_validated, hard_validated = self._preprocess_limit(soft), self._preprocess_limit(hard) # Derive limits from cl-quota.dat according to the limits inheritance rules soft_resolved, hard_resolved = self._combine_user_limits(uid=uid, soft=soft_validated, hard=hard_validated) # Convert limit to format acceptable by setquota utility soft_converted, hard_converted = self._convert_for_sys_utility(soft=soft_resolved, hard=hard_resolved) # Get data from repquota utility cached = self._get_current_quotas(uid)[uid] # Run cmd only if quota changed or force_save is True # If force_save is True it equals to --save-all-paramters in cloudlinux-limits if (soft_converted, hard_converted) != (cached["inodes_soft"], cached["inodes_hard"]) or force_save: # Don't apply limits to root user if uid != '0': device = self._get_home_device(self._fetch_homedir(uid)) if only_store: stdin = f'{uid} {cached["bytes_soft"]} {cached["bytes_hard"]} {soft_converted} {hard_converted}\n' self._device_quota[device] = self._device_quota.get(device, '') + stdin print("Your quota limits have been stored only successfully") else: cmd = [ QuotaWrapper.SETQUOTA, '-u', uid, cached['bytes_soft'], cached['bytes_hard'], soft_converted, hard_converted, device, ] run_command(cmd) print("Your quota limits have been stored successfully") self._sync_quota_files(device) print(f"Your quota limits for device {device} have been applied successfully.") if save: soft_user_dat, hard_user_dat = self._get_user_limits_to_save( uid, soft_validated, hard_validated, force_save=force_save) self._save_user_limits(uid, soft_user_dat, hard_user_dat) print("Quota limits have been stored successfully.") # Always update cl-quota.dat in case both zeroes or both unlimited are provided if (soft_validated == '0' and hard_validated == '0') or (soft_validated == '-1' and hard_validated == '-1'): self._save_user_limits(uid, soft_validated, hard_validated) print("Your quota limits have been stored successfully") if uid == '0': self._apply_all_limits() print("Global quota limits have been applied successfully.") def set_package_limit( self, package: str, soft: Optional[str] = None, hard: Optional[str] = None, save: bool = True, only_store: bool = False ) -> None: """ Sets limits for package :param package: package name :param soft: soft limit value :param hard: hard limit value :param save: save limits to cl-quota.dat :param only_store: store limits in memory, but do not apply them :return: None """ self._check_admin() # Validate limits value, convert 'unlimited' --> '-1', 'default' --> '0' soft_validated, hard_validated = self._preprocess_limit(soft), self._preprocess_limit(hard) # Set limits for empty reseller package if save is True \ and package in self._get_package_quotas(all_packages=True) \ and package not in self._get_package_to_users_map(): # Drive limits from cl-quota.dat according to the limits inheritance rules soft_resolved, hard_resolved = self._get_saved_package_limits_if_none( package, soft_validated, hard_validated) self._save_package_limits(package, soft_resolved, hard_resolved) return if not self._check_package_exists(package): return # Example: {'/dev/sda1': ['502', '504', '515', '521', '501']} device_user_map = self._get_device_user_map() cached_quotas = self._get_current_quotas() for device in device_user_map.keys(): std_in = [] for uid in self._get_package_to_users_map(package): if uid not in device_user_map[device]: continue soft_resolved, hard_resolved = self._combine_package_limits( uid=uid, package=package, soft=soft_validated, hard=hard_validated) soft_converted, hard_converted = self._convert_for_sys_utility(soft=soft_resolved, hard=hard_resolved) try: soft_cached, hard_cached = cached_quotas[uid]['inodes_soft'], cached_quotas[uid]['inodes_hard'] if (soft_converted, hard_converted) != (soft_cached, hard_cached): std_in.append( f'{uid} ' f'{cached_quotas[uid]["bytes_soft"]} {cached_quotas[uid]["bytes_hard"]} ' f'{soft_converted} {hard_converted}' ) except KeyError: pass # skip error when quota is on but not configured if len(std_in) != 0: std_in = ('\n'.join(std_in) + '\n') self._device_quota[device] = self._device_quota.get(device, '') + std_in if save: soft_data, hard_data = self._get_package_limits_to_save(package, soft_validated, hard_validated) self._save_package_limits(package, soft_data, hard_data) if not only_store: self._flush_device_quota() def synchronize(self): """ Read limits from file and applies them to packages and users """ self._check_admin() package_limits = self._get_all_packages_with_limits(clean_dead_packages=True) for package, (soft, hard) in package_limits.items(): self.set_package_limit(package, soft, hard, save=False, only_store=True) self._remove_unexisting_users() self._flush_device_quota() def save_user_cache(self): """ Caches the limits to non-privileged user to see them """ self._check_admin() # get data from repquota utility current_quotas = self._get_current_quotas() # form 2d array for writing to file cache_content = [ [k] + [current_quotas[k][field] for field in self._fields] for k in sorted(current_quotas.keys(), key=int) ] self._get_global_lock(True) file_handler = self._prepare_writer(QuotaWrapper.CACHEFILE) csv_out = csv.writer(file_handler, quoting=csv.QUOTE_MINIMAL) csv_out.writerows(cache_content) self._end_writer(QuotaWrapper.CACHEFILE) self._release_lock() def _apply_all_limits(self, skip_root: bool = True): """Set limits for all users. Skip root user if skip_root is True""" for uid in self._get_uid_to_packages_map().keys(): if uid == '0' and skip_root: continue self.set_user_limit(uid, soft=None, hard=None, save=False, only_store=True) self._flush_device_quota() def _flush_device_quota(self): """Write all device quotas to disk""" quotas_written: bool = False for device in self._device_quota.keys(): cmd = [QuotaWrapper.SETQUOTA, '-bu', device] run_command(cmd, std_in=self._device_quota[device]) quotas_written = True if quotas_written: self._sync_quota_files() self._device_quota = {} def _sync_quota_files(self, device: str \| None = None): """ In order to sync inodes limits in kernel with limits in user.quota file run `quotasync` command. Otherwise `repquota` called right after `setquota` may return old limits existed before `setquota` call. Skipped on the XFS filesystem because XFS does not require a separate quota synchronization step as it handles these operations in real-time. Additionally, the specific functionality required by `quotasync` to sync the disk quota information is not implemented, resulting in an error. """ if device is not None: fs_type = get_filesystem_type(device) if fs_type.lower() == 'xfs': return cmd = [QuotaWrapper.QUOTASYNC, device] else: cmd = [QuotaWrapper.QUOTASYNC, '-a'] run_command(cmd) def _remove_unexisting_users(self): """Remove all records from cl-quota.dat for users which do not exist""" if self._dh.has_section('users'): for uid in self._dh.options('users'): try: # Check user presence self._fetch_homedir(uid) except NoSuchUserException: self._dh.remove_option('users', uid) self._write_data() def _check_package_exists(self, package): """Check whether package exists""" try: self._get_package_to_users_map(package) except NoSuchPackageException: return False else: return True def _get_user_limits_to_save( self, uid: str, soft_validated: Optional[str], hard_validated: Optional[str], force_save: bool = False, ) -> Tuple[Optional[str], Optional[str]]: """ Derive package limit values to save to cl-quota.dat If None passed as limit to method, then replace it by the user's value from cl-quota.dat Update cl-quota.dat only if the derivation result changes """ soft_user_dat, hard_user_dat = self._get_user_limits(uid=uid) soft_resolved, hard_resolved = self._combine_user_limits(uid=uid, soft=soft_validated, hard=hard_validated) soft_none_resolved, hard_none_resolved = self._combine_user_limits(uid=uid, soft=None, hard=None) if soft_resolved != soft_none_resolved or (force_save and soft_validated is not None): soft_user_dat = soft_validated if hard_resolved != hard_none_resolved or (force_save and soft_validated is not None): hard_user_dat = hard_validated return soft_user_dat, hard_user_dat def _get_package_limits_to_save( self, package: str, soft_validated: Optional[str], hard_validated: Optional[str]) -> Tuple[str, str]: """Derive package limit values to save to cl-quota.dat""" p_soft, p_hard = self._get_package_limits(package=package) # If new value is provided, then update package limit p_soft = soft_validated if soft_validated is not None else p_soft p_hard = hard_validated if hard_validated is not None else p_hard return p_soft, p_hard def _check_present_panel(self): """ Return True if control panel present """ if self._panel_present is None: self._panel_present = 'Unknown' != run_command(['/usr/bin/cldetect', '--detect-cp-nameonly']).rstrip() return self._panel_present def _check_admin(self): ''' Raise exception if no admin user ''' if self._euid != 0: raise InsufficientPrivilegesException() def _get_saved_data_handler(self) -> ConfigParser.ConfigParser: ''' Gets ConfigParser handler for future use Loads saved quotas from /etc/container/cl-quotas.dat file ''' self._get_global_lock(True) dh = ConfigParser.ConfigParser(interpolation=None, strict=False) dh.optionxform = str try: dh.read(QuotaWrapper.DATAFILE) except ConfigParser.ParsingError as e: raise MalformedConfigException(e) finally: self._release_lock() return dh def _get_device_user_map(self): """ Returns dictionary mapping devices to lists of users """ if self._device_user_map is not None: return self._device_user_map devices_map = {} device_user_pairs = [] for uid in self._get_list_of_uids(): try: device_user_pairs.append((self._get_home_device(self._fetch_homedir(uid)), uid)) except KeyError: continue for pair in device_user_pairs: if pair[0] not in devices_map: devices_map[pair[0]] = [] devices_map[pair[0]].append(pair[1]) self._device_user_map = devices_map return self._device_user_map def _check_limit(self, limit: Optional[str]) -> Optional[str]: if limit is None or limit == '-1': return limit limit_pattern = re.compile(r'(\d+)') pattern_match = limit_pattern.search(limit) if not pattern_match: raise IncorrectLimitFormatException(limit) return pattern_match.group(1) def _combine_user_limits(self, uid: str, soft: Optional[str] = None, hard: Optional[str] = None) -> Tuple[str, str]: """ Determines user limits by resolving them according to the limits inheritance rule: user limits ---(overridden by provided as method param if provided not None)---> ---> package limits ---> root user (uid=0) limits uid: user id soft: Optional[str] = None: limit value, can be: * None -- value not passed * "-1" -- unlimited * "0" -- default (next value from hierarchy should be considered) * "1", "2", ... -- precice limit values hard: soft: Optional[str] = None: limit value, values range the same as for soft return: Tuple[str, str]: (soft_limit, hard_limit), they can be: * "-1" -- unlimited * "1", "2", ... -- precice limit values """ if uid == '0': soft, hard = self._get_user_limits_override_none(uid=uid, soft=soft, hard=hard) soft, hard = (soft if soft != '0' else '-1', hard if hard != '0' else '-1') return soft, hard soft, hard = self._get_user_limits_override_none(uid=uid, soft=soft, hard=hard) for package in self._get_uid_to_packages_map(uid): soft, hard = self._get_package_limits_override_default(package=package, soft=soft, hard=hard) soft, hard = self._get_user_limits_override_default(uid='0', soft=soft, hard=hard) soft, hard = (soft if soft != '0' else '-1', hard if hard != '0' else '-1') return soft, hard def _get_user_limits_override_none(self, uid: str, soft: Optional[str] = None, hard: Optional[str] = None) -> Tuple[str, str]: """Get user limits from cl-quota.dat. If limit is None, then override it by user's limit :param str uid: user id :param Optional[str] soft: limit value, can be: * None -- value not passed, should be overridden by user's limit * "-1" -- unlimited * "0" -- default (next value from hierarchy should be taken) * "1", "2", ... -- values :param Optional[str] hard: limit values same as for soft :return Tuple[str, str]: derived limits """ user_soft, user_hard = self._get_user_limits(uid=uid) # Override by passed limits values soft = self._check_limit(limit=soft if soft is not None else user_soft) hard = self._check_limit(limit=hard if hard is not None else user_hard) return soft, hard def _get_user_limits_override_default(self, uid: str, soft: str, hard: str) -> Tuple[str, str]: """Get user limits from cl-quota.dat. If limit is default, then override it by user's limit :param str uid: user id :param Optional[str] soft: limit value, can be: * "-1" -- unlimited * "0" -- default (next value from hierarchy should be taken) * "1", "2", ... -- values :param Optional[str] hard: limit values same as for soft :return Tuple[str, str]: derived limits """ user_soft, user_hard = self._get_user_limits(uid=uid) # Override by passed limits values soft = self._check_limit(limit=soft if soft != '0' else user_soft) hard = self._check_limit(limit=hard if hard != '0' else user_hard) return soft, hard def _get_user_limits(self, uid: str) -> Tuple[str, str]: """Try to get user's limits from cl-quota.dat""" try: user_soft, user_hard = self._dh.get('users', uid).split(':') except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): user_soft, user_hard = '0', '0' return user_soft, user_hard def _get_package_limits_override_default(self, package: str, soft: str, hard: str) -> Tuple[str, str]: """Get package limits from cl-quota.dat. If passed limit is default, then override it by package's limit""" pack_soft, pack_hard = self._get_package_limits(package=package) soft = self._check_limit(limit=pack_soft if soft == '0' else soft) hard = self._check_limit(limit=pack_hard if hard == '0' else hard) return soft, hard def _get_package_limits(self, package: str) -> Tuple[str, str]: """Try to get package's limits from cl-quota.dat""" try: soft, hard = self._get_package_from_dh(package) except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): soft, hard = '0', '0' return soft, hard @staticmethod def _convert_for_sys_utility(soft: Optional[str], hard: Optional[str]) -> Tuple[Optional[str], Optional[str]]: """Converts limits for setquota utility which threats 0 as unlimited""" if soft == '-1': soft = '0' if hard == '-1': hard = '0' return soft, hard def _combine_package_limits( self, package: str, uid: str, soft: Optional[str], hard: Optional[str]) -> Tuple[str, str]: """ Determines package limits taking into account saved user and default ones """ u_soft, u_hard = self._get_user_limits(uid=uid) if IS_DA and is_quota_inheritance_enabled(): # Check the real user's package and save his quotas (instead of setting `DEFAULT` package ones) # This is only DA's specificall da_real_package = self._get_da_real_package(uid=uid) if da_real_package != package: da_real_quotas = self._get_package_quotas(packname=da_real_package, all_packages=True) soft = p_soft = da_real_quotas[da_real_package]['inodes_soft'] hard = p_hard = da_real_quotas[da_real_package]['inodes_hard'] else: p_soft, p_hard = self._get_package_limits(package=package) else: p_soft, p_hard = self._get_package_limits(package=package) # Combine package limits with new package limits # If user limit is non-default, take it, otherwise use package limit # Override package limit by the new one if it's possible soft = u_soft if u_soft != '0' else (soft if soft is not None else p_soft) hard = u_hard if u_hard != '0' else (hard if hard is not None else p_hard) # If package limits absent, use default limits soft, hard = self._get_user_limits_override_default(uid='0', soft=soft, hard=hard) soft, hard = (soft if soft != '0' else '-1', hard if hard != '0' else '-1') return soft, hard def _get_saved_package_limits_if_none(self, package, soft=None, hard=None): """ Applies saved package limits if none has been passed """ try: pack_soft, pack_hard = self._get_package_from_dh(package) if soft is None and pack_soft != '0': soft = pack_soft if hard is None and pack_hard != '0': hard = pack_hard except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): pass soft = self._check_limit(soft) hard = self._check_limit(hard) return soft, hard def _get_da_real_package(self, uid: str) -> str: """ Get real package name for DA user :param str uid: user id :return str: retrieved package name """ username = ClPwd().get_names(int(uid))[0] return clcontrollib.DirectAdmin()._get_user_package(username) def _get_current_quotas(self, uid=None): """ Retrieves current quotas. If euid == 0, use data from repquota utility, else from /etc/container/cl-quotas.cache file """ if self._euid != 0: return self._load_user_cache() if not self._quota: # Retrieves quotas from repquota utility self._quota = self._load_current_quotas() if uid: try: return {uid: self._quota[uid]} except KeyError: self._check_if_quota_enabled(uid) raise NoSuchUserException(uid) return self._quota def _get_package_quotas(self, packname=None, all_packages=False): """ Prepares package limits data for outputting (call only from get_package_limits/get_all_packages_limits - main) :param packname: Package name for get limits. If present, function returns limits only for this package, else - all packages :param all_packages: If False reads only used and admin's packages, True - all packages (including reseller packages without users) :return Dictionary of package limits: {package_name: {'inodes_used': 'xxx', 'inodes_soft': 'yyy', 'inodes_hard': 'zzz'} """ q = {} if all_packages: # Get list of all packages list_of_packages = self._get_all_package_list() else: # Get list of used packages + all admin's packages list_of_packages = self._get_list_of_packages() for package in list_of_packages: values = ['-'] try: if package == VE_DEFAULT_PACKAGE: # Because "VE_DEFAULT" package is not a real package and just # uses limits from LVE == 0 we should read it's limits # from there soft, hard = self._dh.get('users', '0').split(':') else: soft, hard = self._dh.get('packages', package).split(':') soft = self._check_limit(soft) hard = self._check_limit(hard) if soft == '-1': soft = '-' if hard == '-1': hard = '-' values.extend([soft, hard]) except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): values.extend(['0', '0']) q.update(self._populate(package, values)) if packname: try: return {packname: q[packname]} except KeyError: raise NoSuchPackageException(packname) return q def _populate(self, item, data): return {item: dict(list(map((lambda x: (x[1], data[x[0]])), enumerate(self._fields[3:]))))} def _get_list_of_packages(self): return list(self._get_package_to_users_map().keys()) def _get_list_of_uids(self): return list(self._get_uid_to_packages_map().keys()) def _get_package_to_users_map(self, package=None): if not self._package_to_uids_map: self._package_to_uids_map = self._load_package_uids_data() if package: try: return self._package_to_uids_map[package] except KeyError: raise NoSuchPackageException(package) return self._package_to_uids_map def _check_if_quota_enabled(self, uid): if uid in self._quota_enabled_list: return home_dir = self._fetch_homedir(uid) quota_disabled_message = check_quota_enabled(path=home_dir) if quota_disabled_message: raise UserQuotaDisabledException(uid=uid, homedir=home_dir, message=quota_disabled_message) else: self._quota_enabled_list.append(uid) def _get_uid_to_packages_map(self, uid=None): if not self._uid_to_packages_map: self._package_to_uids_map = self._load_package_uids_data() if uid: try: return self._uid_to_packages_map[uid] except KeyError: raise NoSuchUserException(uid) return self._uid_to_packages_map def _get_packages_uids_from_cpapi(self) -> Dict[str, List[str]]: """ Retrieve package-uids map from cpapi. Only for custom panels. See LU-610 for details. Null packages coming from cpapi are considered to be 'default' package. :return: Dictionary with data. Example response: {'default': ['1038', '1043', '1046'], 'res1_pack1': ['1044'], 'pack1': ['1042']} Coorresponding self._uid_to_packages_map value: {'1038': ['default'], '1042': ['pack1'], '1043': ['default'], '1044': ['res1_pack1'], '1046': ['default']} """ try: users_packages = list_users() except (OSError, CPAPIExternalProgramFailed, EncodingError) as e: raise ExternalProgramFailed('%s. Can not get users' % (str(e))) # Example of users_packages: # {1000: {'reseller': 'root', 'package': 'Package1'}, # 1001: {'reseller': 'res1', 'package': 'BusinessPackage'}, # 1002: {'reseller': 'root', 'package': None}} packages_users = defaultdict(list) # type: Dict[str, List[str]] self._uid_to_packages_map = defaultdict(list) # type: Dict[str, List[str]] for uid, uid_data in users_packages.items(): s_uid = str(uid) package = uid_data['package'] if uid_data['package'] is not None else VE_DEFAULT_PACKAGE packages_users[package].append(s_uid) self._uid_to_packages_map[s_uid].append(package) try: admin_pkgs = admin_packages(raise_exc=True) except (OSError, CPAPIExternalProgramFailed) as e: raise ExternalProgramFailed('%s. Can not get admin packages' % (str(e))) for package in admin_pkgs: packages_users.setdefault(package if package is not None else VE_DEFAULT_PACKAGE, []) packages_users.setdefault(VE_DEFAULT_PACKAGE, []) return packages_users def _load_package_uids_data(self) -> Dict[str, List[str]]: """ Gets map of packages and users :rtype dict :return Dictionary with data. Example: {'default': ['1038', '1043', '1046'], 'res1_pack1': ['1044'], 'pack1': ['1042']} """ packages = {} if self._euid != 0: return packages # if packages not supported all user has 'VE_DEFAULT' package if not self._check_present_panel(): packages[VE_DEFAULT_PACKAGE] = list(map(str, _get_users_list())) self._uid_to_packages_map = {i: [VE_DEFAULT_PACKAGE] for i in packages[VE_DEFAULT_PACKAGE]} return packages return self._get_packages_uids_from_cpapi() def _get_all_package_list(self): """ Retrives all (root and resellers) panel package list :return: List of package names """ # If list already loaded - do nothing if self._all_package_list: return self._all_package_list try: self._all_package_list = [] list_admin_packages = admin_packages(raise_exc=True) for package in list_admin_packages: self._all_package_list.append(package) except (OSError, CPAPIExternalProgramFailed) as e: raise ExternalProgramFailed('%s. Can not get admin packages' % (str(e))) try: dict_resellers_packages = resellers_packages(raise_exc=True) for packages_list in dict_resellers_packages.values(): for package in packages_list: self._all_package_list.append(package) except (OSError, CPAPIExternalProgramFailed) as e: raise ExternalProgramFailed('%s. Can not get reseller packages' % (str(e))) # Add 'VE_DEFAULT' package to list if VE_DEFAULT_PACKAGE not in self._all_package_list: self._all_package_list.append(VE_DEFAULT_PACKAGE) return self._all_package_list def _convert_data_to_tuples(self, data): ''' Convert dict to tuples for passing to printing routines ''' for key in data.keys(): try: entry = tuple(map((lambda x: (x, data[key][x])), self._fields[3:])) data[key] = entry except KeyError: continue return data def _load_current_quotas(self): """ Gets current quota settings from repqouta utility for further processing """ q = {} device = None devices = self._devices cmd = [QuotaWrapper.REPQUOTA, '-una'] data = run_command(cmd) grace_regex_pattern = re.compile(r'(block\|inode)\sgrace\stime:?\s(\d[\w:]+)(?:;\|$\|\s)', re.IGNORECASE) for line in data.splitlines(): if line.startswith('#'): if not device: continue parts = line.split() if len(parts) != 8: parts = self._remove_redundant_fields_from_input(parts) uid = parts[0][1:] if uid == '0': # We do not want to limit root :) continue try: if device not in devices: device = self._find_unknown_device(device) if device in devices and self._is_home_device(self._fetch_homedir(uid), device): q[uid] = dict(list(map((lambda x: (self._fields[x[0]], x[1])), enumerate(parts[2:])))) except (KeyError, IndexError, NoSuchUserException): continue elif line.startswith('***'): device = line[line.find('/dev'):].strip() elif 'grace' in line: found = grace_regex_pattern.findall(line) if found: self._grace.update(dict(list(map((lambda x: (x[0].lower(), x[1])), found)))) q.update(self._add_default()) return q def _remove_redundant_fields_from_input(self, parts): stripped_parts = parts[:2] is_digit_pattern = re.compile(r'^\d+$') stripped_parts.extend( [field for field in parts[2:] if is_digit_pattern.search(field)]) return stripped_parts def _fetch_homedir(self, uid): if len(self._uid_to_homedir_map) == 0: self._uid_to_homedir_map.update({str(entry.pw_uid): entry.pw_dir for entry in pwd.getpwall()}) try: return self._uid_to_homedir_map[uid] except KeyError: raise NoSuchUserException(uid) def _load_quota_devices(self): """ Gets mounted filesystems list and picks ones with quota on Example of returned data structure: {'/dev/mapper/VolGroup-lv_root': [ {'mountpoint': '/', 'quota_file': 'quota.user', 'quota_type': 'vfsv0'}, {'mountpoint': '/var', 'quota_file': 'quota.user', 'quota_type': 'vfsv0'} ], '/dev/mapper/VolGroup-lv_root2': [ {'mountpoint': '/', 'quota_file': 'quota.user', 'quota_type': 'vfsv0'}, {'mountpoint': '/var', 'quota_file': 'quota.user', 'quota_type': 'vfsv0'} ] } """ devices = {} # type: Dict[str, List[Dict[str, str]]] proc_mounts_stream = open(QuotaWrapper.PROC_MOUNTS) split_patt = re.compile(r' \|,') for line in proc_mounts_stream: if line.startswith('rootfs /'): continue line_splited = split_patt.split(line) device = line_splited[0] mountpoint_data = {'mountpoint': line_splited[1]} for line_splited_element in line_splited: if line_splited_element.startswith('usrquota=') or line_splited_element.startswith('usruota='): mountpoint_data['quota_file'] = line_splited_element.split('=')[1] elif line_splited_element.startswith('jqfmt='): mountpoint_data['quota_type'] = line_splited_element.split('=')[1] if device in devices: devices[device].append(mountpoint_data) else: devices[device] = [mountpoint_data] proc_mounts_stream.close() if len(devices) == 0: # TODO: this only can happen when system HAS NO MOUNTS AT ALL raise QuotaDisabledException() return devices def _load_user_cache(self): ''' For non-privileged user we outputting data from the file ''' q = {} try: self._get_global_lock() fo = open(QuotaWrapper.CACHEFILE) cvs_in = csv.reader(fo, delimiter=',') except (OSError, IOError): # We don't want to confuse a panel with error messages. # Let the data be zeroes until they arrive return {str(self._euid): dict.fromkeys(self._fields, '0')} finally: self._release_lock() uid = str(self._euid) for row in cvs_in: if row[0] == uid: q.update({row[0]: dict(list(map( (lambda x: (self._fields[x], row[x+1])), range(len(self._fields)))))}) # pylint: disable=range-builtin-not-iterating break # We want to prevent crazy cases like misedited cache file if not q: return {str(self._euid): dict.fromkeys(self._fields, '0')} return q def _get_mountpoint_device_map(self, devices) -> List[Tuple[str, str]]: """ return list tuple ('mountpoin tpath', 'device') reverse sorted by deep mountpoint path [('/mountpoint_path/path', '/device'), ('/mountpoint_path', '/device')] """ def sort_by_deep_path(device_mountpoint): if device_mountpoint[0] == '/': deep_path = 0 else: deep_path = device_mountpoint[0].count('/') return deep_path mountpoint_device_map = [] for device, mountpoint_data_list in devices.items(): for mountpoint_data in mountpoint_data_list: mountpoint_path = mountpoint_data['mountpoint'] mountpoint_device_map.append((mountpoint_path, device)) mountpoint_device_map.sort(key=sort_by_deep_path, reverse=True) return mountpoint_device_map def _get_home_device(self, home): """ Returns device user homedir is on """ def _add_slash(path): if path and path[-1] != '/': path += '/' return path dirname = _add_slash(os.path.dirname(home)) for mounpoint_path, device in self._mountpoint_device_mapped: if dirname.startswith(_add_slash(mounpoint_path)): return device def _is_home_device(self, home, device): """ Checks if a device is user homedir device """ return self._get_home_device(home) == device def _find_unknown_device(self, device): try: dev = os.stat(device)[ST_DEV] dev_to_find = (os.major(dev), os.minor(dev)) for current_device in self._devices.keys(): dev = os.stat(current_device)[ST_DEV] if dev_to_find == (os.major(dev), os.minor(dev)): return current_device except OSError: return device def _add_default(self): """ Insert 'default' quota. Calls only from _load_current_quotas, after parsing repquota's output """ values = ['-', '0', '0', '-'] try: user_soft, user_hard = self._dh.get('users', '0').split(':') # Replace -1 to 0 for set unlimited limit if user_soft == '-1': user_soft = '0' if user_hard == '-1': user_hard = '0' values.extend([user_soft, user_hard]) except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): values.extend(['0', '0']) return {'0': dict(list(map((lambda x: (x[1], values[x[0]])), enumerate(self._fields))))} def _save_user_limits(self, uid, soft, hard): """ Saves user limits """ if soft is None: soft = '0' if hard is None: hard = '0' # remove user limits if they are equal to default if soft == '0' and hard == '0' and self._dh.has_section('users'): self._dh.remove_option('users', uid) else: if not self._dh.has_section('users'): self._dh.add_section('users') self._dh.set('users', uid, '%s:%s' % (soft, hard)) self._write_data() def _save_package_limits(self, package, soft, hard): """ Saves package limits """ if soft is None: soft = '0' if hard is None: hard = '0' if soft == '0' and hard == '0' and self._dh.has_section('packages'): self._dh.remove_option('packages', package) else: if not self._dh.has_section('packages'): self._dh.add_section('packages') self._dh.set('packages', package, '%s:%s' % (soft, hard)) self._write_data() self._copy_package_limits_to_cpanel(package) def _copy_package_limits_to_cpanel(self, package): """ Copy package quota limits from cl-quotas.dat to cpanel packages data """ if not cldetectlib.is_cpanel(): return # skip func if panel not cPanel package_path = f'/var/cpanel/packages/{package}' cpanel_package_lines = get_file_lines(package_path) if len(cpanel_package_lines) == 0: return # skip func if no cPanel package found old_cpanel_data, modified_cpanel_lines = self._parse_cpanel_package_data(cpanel_package_lines) if old_cpanel_data is None and modified_cpanel_lines is None: return # skip func if no lve extension in package # don't rewrite cpanel package file if new quotas for package are the same quotas_data = self._get_package_quotas(package, all_packages=True)[package] # unlimited quotas for package are indicated as '-', # but in package we want to write '-1' for key, value in quotas_data.items(): if value == '-': quotas_data[key] = '-1' old_cpanel_limits = (old_cpanel_data.get('inodes_soft', '0'), old_cpanel_data.get('inodes_hard', '0')) current_quota_limits = (quotas_data['inodes_soft'], quotas_data['inodes_hard']) if old_cpanel_limits == current_quota_limits: return for limit_type in ('inodes_soft', 'inodes_hard'): limit_string = 'lve_' + str(limit_type) + '=' + str(quotas_data[limit_type]) + '\n' modified_cpanel_lines.append(limit_string) write_file_lines(package_path, modified_cpanel_lines, 'w') @staticmethod def _parse_cpanel_package_data(cpanel_package_lines): """ Process cpanel_package_lines - get values of all old lve_ limits and remove lines with limits that would be changed """ cpanel_package_lines_modified = cpanel_package_lines[:] old_cpanel_data = {} for line in cpanel_package_lines: if line.startswith('lve_'): line_parts = line.strip().split('=') limit_name = line_parts[0].replace('lve_', '').strip() if line_parts[1] != 'DEFAULT': old_cpanel_data[limit_name] = line_parts[1] if limit_name in ('inodes_soft', 'inodes_hard'): cpanel_package_lines_modified.remove(line) if line.startswith('_PACKAGE_EXTENSIONS') and 'lve' not in line: return None, None return old_cpanel_data, cpanel_package_lines_modified def _save_data(self, soft, hard, item, item_type): ''' Saves data to a file ''' if soft == '0' and hard == '0': try: self._dh.remove_option(item_type, item) except ConfigParser.NoSectionError: pass else: if not self._dh.has_section(item_type): self._dh.add_section(item_type) self._dh.set(item_type, item, '%s:%s' % (soft, hard)) self._write_data() def _prepare_writer(self, filepath): """ Open temporary file for writing and return file object """ path = os.path.dirname(filepath) try: fd, temp_path = tempfile.mkstemp(prefix='lvetmp_', dir=path) file_handler = os.fdopen(fd, 'w') self._tmp = temp_path return file_handler except (IOError, OSError): if os.path.exists(temp_path): os.unlink(temp_path) raise GeneralException("Could not save data") def _end_writer(self, path): ''' Routines after writing to file ''' try: mask = S_IRUSR \| S_IWUSR \| S_IRGRP \| S_IROTH os.rename(self._tmp, path) os.chmod(path, mask) except OSError: pass def _write_data(self): ''' Actual place of saving data to a file ''' self._get_global_lock(True) file_handler = self._prepare_writer(QuotaWrapper.DATAFILE) self._dh.write(file_handler) self._end_writer(QuotaWrapper.DATAFILE) self._release_lock() ########################## # File lock functions def _get_global_lock(self, write=False): if write: QuotaWrapper.LOCK_WRITE = True if QuotaWrapper.LOCK_FD is None: try: QuotaWrapper.LOCK_FD = open(QuotaWrapper.LOCK_FILE, 'r') except (IOError, OSError): raise GeneralException("Can't open lock file for reading") try: fcntl.flock(QuotaWrapper.LOCK_FD.fileno(), fcntl.LOCK_EX) except IOError: raise GeneralException("Can't get lock") def _release_lock(self): if (not QuotaWrapper.LOCK_WRITE) and (QuotaWrapper.LOCK_FD is not None): QuotaWrapper.LOCK_FD.close() QuotaWrapper.LOCK_FD = None	52.39 KB	2026-02-26 15:33
📁 __pycache__	-	2026-03-21 15:50
📄 cl_quota.py #!/opt/cloudlinux/venv/bin/python3 -bb # -- 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 __future__ import print_function from __future__ import division from __future__ import absolute_import import os import sys import re import getopt import pwd import logging import logging.handlers from cldetectlib import get_boolean_param, CL_CONFIG_FILE from clquota import ( QuotaWrapper, NoSuchPackageException, NoSuchUserException, IncorrectLimitFormatException, InsufficientPrivilegesException, GeneralException, QuotaDisabledException ) from clquota.utils import ( print_text, print_csv, print_json, print_text_error, print_csv_error, print_json_error ) import logging logger_d = logging.getLogger(__name__) logger_d.setLevel(logging.DEBUG) # fh = logging.FileHandler('/root/cl_quota.log') # fh.setLevel(logging.DEBUG) # fh.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) # logger_d.addHandler(fh) def check_username_and_convert_to_uid(username=None): ''' Checks if username is valid one and returns UID as string ''' if username.lower() == 'default': return '0' try: uid = pwd.getpwnam(username).pw_uid return str(uid) except KeyError: raise NoSuchUserException(username) def check_uid(uid=None): ''' Checks if UID is valid one and retuns it as string ''' p = re.compile(r'(\d+)') pm = p.search(uid) if not pm: raise NoSuchUserException(uid) uid = pm.group(1) if uid == '0': # this is default. No need to check further return uid try: pwd.getpwuid(int(uid)) except KeyError: raise NoSuchUserException(uid) return uid def setup_system_log(): """ Sets syslog logger and returns its handler """ logger = logging.getLogger('clquota') logger.setLevel(logging.INFO) syslogger = logging.handlers.SysLogHandler(address='/dev/log') formatter = logging.Formatter(fmt='%(name)s:%(levelname)s %(message)s') syslogger.setFormatter(formatter) logger.addHandler(syslogger) return logger def usage(): print('') print('Usage: ' + sys.argv[0] + ' [OPTIONS]') print('If no options given prints quota statistics for all users') print('Options:') print('') print(' -u \| --user : specifies the user') print(' -U \| --user-id : specifies the user ID') print(' -S \| --soft-limit : sets the soft limit for a user. ' 'Pass 0 or \'default\' to set package default limit. Pass -1 or \'unlimited\' to cancel limit') print(' -H \| --hard-limit : sets the hard limit for a user. ' 'Pass 0 or \'default\' to set package default limit. Pass -1 or \'unlimited\' to cancel limit') print(' -V \| --csv : returns data as comma separated values') print(' -J \| --json : returns data as json') print(' -p \| --package : specifies a package to set or get limits') print(' -P \| --package-limits : prints package limits') print(' -a \| --all-package-limits : prints all package limits (including packages without limits)') print(' -Y \| --sync : synchronizes packages and users limits with the database') print(' -C \| --cache-content : cache quota data to a file the database') print(' -F \| --force : save user quotas even when they are equal to defaults') print(' --check : Deprecated. Check if quotas is enabled/activated/suported; ' 'if disabled show diagnostic information; using with --user or --user-id options') def quota_sync(): """ Quota synchronization :return: """ q = QuotaWrapper() q.synchronize() q.save_user_cache() def check_autosync_disabled(): """ Check autosynchronization parameter in config file and exit if it`s disabled :return: """ if not get_boolean_param(CL_CONFIG_FILE, 'cl_quota_limits_autosync', default_val=True): print("cl-quota limits autosynchronization is disabled in config " + CL_CONFIG_FILE) sys.exit(0) def parse_args(argv_): try: opts, args = getopt.getopt(argv_, 'PaVJYCp:u:U:S:H:F', ['package-limits', 'all-package-limits', 'csv', 'json', 'package=', 'user-id=', 'user=', 'soft-limit=', 'hard-limit=', 'sync', 'cache', 'check', 'profiling', 'force']) except getopt.GetoptError: usage() sys.exit(1) config = {} config['soft'] = None config['hard'] = None config['user'] = None config['user-id'] = None config['format'] = 'text' config['check'] = False config['profiling'] = False config['force_save'] = False for o, a in opts: if o in ['-P', '--package-limits']: config['package'] = None elif o in ['-a', '--all-package-limits']: config['all-package'] = None elif o in ['-p', '--package']: config['package'] = a elif o in ['-u', '--user']: config['user'] = a elif o in ['-U', '--user-id']: config['user-id'] = a elif o in ['-S', '--soft-limit']: config['soft'] = a elif o in ['-H', '--hard-limit']: config['hard'] = a elif o in ['-Y', '--sync']: config['sync'] = True elif o in ['-C', '--cache']: config['cache'] = True elif o == '--check': config['check'] = True elif o in ['-V', '--csv']: config['format'] = 'csv' elif o in ['-J', '--json']: config['format'] = 'json' elif o in ['--profiling']: config['profiling'] = True elif o in ['-F', '--force']: config['force_save'] = True return config def main(argv_): output_dispatcher = { 'text': print_text, 'csv': print_csv, 'json': print_json, 'err_text': print_text_error, 'err_csv': print_csv_error, 'err_json': print_json_error } config = parse_args(argv_) try: if 'sync' in config and 'cache' in config: # exit in case synchronization is disabled in config file check_autosync_disabled() # We don't want to do anything if no saved data if not os.path.exists(QuotaWrapper.DATAFILE): return if config['profiling']: # profiler mode profiler_log = 'quota-profiling.log' print("Profiling end; result saved to " + profiler_log) import profile profile.run('quota_sync()', profiler_log) # Profiler result decode import pstats p = pstats.Stats(profiler_log) print('--------------------------------------') print() print('Cumulative time:') # p.sort_stats('cumulative').print_stats(20) p.sort_stats('calls').print_stats(20) print('--------------------------------------') print() print('Total time:') # p.sort_stats('time').print_stats(20) p.sort_stats('calls').print_stats(20) else: # work without profiler quota_sync() return q = QuotaWrapper() if 'package' in config and not (config['soft'] or config['hard']): if config['package'] is None: # Show limits for all used packages output_dispatcher[config['format']](q.get_package_limits(None), title='package') else: # Show limits only for given package output_dispatcher[config['format']](q.get_all_packages_limits(config['package']), title='package') return if 'all-package' in config: # Show all packages output_dispatcher[config['format']](q.get_all_packages_limits(), title='package') return uid = None if config['user']: uid = check_username_and_convert_to_uid(config['user']) elif config['user-id']: uid = check_uid(config['user-id']) if config['check'] and uid is not None and int(uid) != 0: q._check_if_quota_enabled(uid) if config['soft'] or config['hard']: if 'package' in config and config['package'] is not None: # Set package limits q.set_package_limit( package=config['package'], soft=config['soft'], hard=config['hard']) return # Set user's limits q.set_user_limit(uid=uid, soft=config['soft'], hard=config['hard'], force_save=config['force_save']) return if 'sync' in config: q.synchronize() return if 'cache' in config: q.save_user_cache() return if uid: if not config['check']: output_dispatcher[config['format']](q.get_user_limits(uid), title='id', sort_key=int) else: output_dispatcher[config['format']](q.get_all_users_limits(), title='id', sort_key=int) except QuotaDisabledException as e: if config['check']: print(str(e)) sys.exit(1) except (NoSuchPackageException, NoSuchUserException, InsufficientPrivilegesException, IncorrectLimitFormatException, GeneralException) as e: output_dispatcher['err_' + config['format']](e) setup_system_log().info(e) sys.exit(1) except Exception as e: output_dispatcher['err_' + config['format']](e) setup_system_log().error(e) sys.exit(1) if __name__ == '__main__': main(sys.argv[1:])	10.09 KB	2026-02-26 15:33
📄 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 from __future__ import print_function from __future__ import division from __future__ import absolute_import import json import sys import csv import time from clcommon.ui_config import UIConfig from clcommon import FormattedException def print_text(data, title='id', sort_key=str): ''' Prints data as pure text ''' data_keys = list(data.keys()) if len(data_keys) == 0: return fields = list(map((lambda x: x[0]), data[data_keys[0]])) fmt_list = ['%32s'] fmt_list.extend(['%16s'] * (len(fields))) fmt = ''.join(fmt_list) print(fmt % tuple([title] + fields)) print('-' * 80) for item in sorted(data_keys, key=sort_key): row = [item] row.extend(list(map((lambda x:x[1]), data[item]))) print(fmt % tuple(row)) def print_csv(data, title='id', sort_key=str): ''' Prints data as comma separated values ''' data_keys = list(data.keys()) if len(data_keys) == 0: return fields = list(map((lambda x: x[0]), data[data_keys[0]])) fields = [title] + fields csv_out = csv.writer(sys.stdout, quoting=csv.QUOTE_MINIMAL) fmt = ','.join(['%s'] * len(fields)) print(fmt % tuple(fields)) for item in sorted(data_keys, key=sort_key): row = [item] row.extend(list(map((lambda x:x[1]), data[item]))) csv_out.writerow(row) def print_json(data, title='id', sort_key=str): """ Prints data as json """ print(json.dumps({ 'result': 'success', 'timestamp': time.time(), 'items': data, 'quota_show': check_quota_show(), }, indent=1)) def print_json_error(error): """ Prints error in json format that is later used by cl-limits """ if isinstance(error, FormattedException): print(json.dumps({ 'result': error.message, 'context': error.context, 'details': error.details }, indent=1)) else: print(json.dumps({ 'result': str(error) }, indent=1)) def print_text_error(message): print("ERROR: %s" % (message)) print( "Something went wrong while running cl-quota. " "Make sure the utility is installed and working correctly. " "If the problem persists, contact CloudLinux support for help.", end="", ) def print_csv_error(message): csv_out = csv.writer(sys.stdout, quoting=csv.QUOTE_ALL) csv_out.writerow(["ERROR", message]) def check_quota_show(): """ Show quota enabled :return: bool """ return UIConfig().get_param('showUserInodesUsage', 'inodeLimits') is True	2.76 KB	2026-02-26 15:33

Terminal