Graybyt3 Was Here

Linux server6.maghost.ro 4.18.0-477.21.1.lve.1.el8.x86_64 #1 SMP Tue Sep 5 23:08:35 UTC 2023 x86_64
Apache
91.213.11.6

File Name : __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 builtins import map
from collections import defaultdict
from stat import S_IRGRP, S_IROTH, S_IRUSR, S_IWUSR, ST_DEV

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, run_command, write_file_lines

IS_DA = clcontrollib.detect.is_da()
DEFAULT_PACKAGE = 'default'

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
    """
    PROC_MOUNTS = '/proc/mounts'
    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 = {}
        self._uid_to_packages_map = {}
        self._uid_to_homedir_map = {}
        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):

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_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._dh.get('packages', package).split(':')
                # 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, soft=None, hard=None, save=True, sync=True,
                       force_save=False, only_store=False):
        """
        Sets limits for users
        :return: None
        """
        # if sync is False both limits should be provided
        if not sync and (soft is None or hard is None):
            return
        self._check_admin()
        soft_arg = soft
        hard_arg = hard
        # convert 'unlimited' --> '-1', 'default' --> '0'
        # and check other limits values by calling _check_limit function
        soft = self._preprocess_limit(soft_arg)
        hard = self._preprocess_limit(hard_arg)
        if uid == '0':
            # Replace -1 to 0 for set unlimited limit
            if soft == '-1':
                soft = '0'
            if hard == '-1':
                hard = '0'
            # Use clquota.dat limits if they not provided
            if self._dh.has_section('users') and self._dh.has_option('users', '0'):
                # uid present in clquota.dat
                limits = self._dh.get('users', '0').split(':')
                cache_soft = soft or limits[0]
                cache_hard = hard or limits[1]
            else:
                cache_soft = soft or '0'
                cache_hard = hard or '0'
            # Set limits for all non-package users.
            self._apply_to_all_if_not_set(soft=cache_soft, hard=cache_hard)

# Save cl-quota.dat
            if save:
                # if limit set as -1 or unlimited save it to file as -1
                # else save fact value
                if soft_arg in ['-1', 'unlimited']:
                    soft = '-1'
                elif soft is None:
                    soft = cache_soft
                if hard_arg in ['-1', 'unlimited']:
                    hard = '-1'
                elif hard is None:
                    hard = cache_hard
                self._save_user_limits(uid='0', soft=soft, hard=hard)

# Process packages with '0' limits
            package_limits_dict = self._get_all_packages_with_limits()
            for package in package_limits_dict:
                p_soft, p_hard = package_limits_dict[package]
                if p_soft == '0' or p_hard == '0':
                    self.set_package_limit(package, p_soft, p_hard, save, sync, only_store=only_store)
            return
        if sync:
            user_combine_soft, user_combine_hard = self._combine_user_limits(uid=uid, soft=soft, hard=hard)
        else:
            user_combine_soft, user_combine_hard = '0', '0'
        user_soft, user_hard = user_combine_soft, user_combine_hard
        # Replace -1 to 0 for set unlimited limit
        if soft == '-1' or user_combine_soft == '-1':
            user_soft = '0'
        if hard == '-1' or user_combine_hard == '-1':
            user_hard = '0'
        # get data from repquota utility, or from /etc/container/cl-quotas.cache file
        saved_quotas = self._get_current_quotas(uid)
        cached = saved_quotas[uid]
        # if force_save is True it equals to --save-all-paramters in cloudlinux-limits
        if cached["inodes_hard"] != user_hard or cached["inodes_soft"] != user_soft or force_save:
            # run cmd only if quota changed
            user_package = self._get_uid_to_packages_map(uid)[0]
            if user_package == DEFAULT_PACKAGE:
                # Use clquota.dat limits if they not provided
                if self._dh.has_section('users') and self._dh.has_option('users', uid):
                    # uid present in clquota.dat
                    limits = self._dh.get('users', uid).split(':')
                    cache_soft = limits[0]
                    cache_hard = limits[1]
                else:
                    # uid absent in clquota.dat
                    cache_soft = '0'
                    cache_hard = '0'
                soft_limit = user_soft or cache_soft
                hard_limit = user_hard or cache_hard
            else:
                # User not in default package, use limits from combined
                soft_limit = user_soft
                hard_limit = user_hard

cmd = [
                QuotaWrapper.SETQUOTA,  # /usr/sbin/setquota
                '-u', uid,
                cached['bytes_soft'],
                cached['bytes_hard'],
                soft_limit,
                hard_limit,
                self._get_home_device(self._fetch_homedir(uid))
            ]
            run_command(cmd)
            if save:
                if user_combine_soft == '-1':
                    soft_limit = '-1'
                else:
                    if soft in ['0', '-1']:
                        soft_limit = soft
                if user_combine_hard == '-1':
                    hard_limit = '-1'
                else:
                    if hard in ['0', '-1']:
                        hard_limit = hard
                self._save_user_limits(uid, soft_limit, hard_limit)
        if (soft == '0' and hard == '0') or (soft == '-1' and hard == '-1'):
            self._save_user_limits(uid, soft, hard)

def set_package_limit(self, package, soft=None, hard=None, save=True, sync=True, only_store=False):
        """
        Sets limits for package
        :rtype : None
        """
        clpwd = ClPwd()
        # 'default' package not processing
        if package == DEFAULT_PACKAGE:
            return
        # if sync is False both limits should be provided
        if not sync and (soft is None or hard is None):
            return
        self._check_admin()
        # convert 'unlimited' --> '-1', 'default' --> '0'
        # and check other limits values by calling _check_limit functon
        soft = self._preprocess_limit(soft)
        hard = self._preprocess_limit(hard)
        std_in = []
        if sync:
            soft, hard = self._get_saved_package_limits_if_none(package, soft, hard)
        # Set limits for empty reseller package
        if save and package in self._get_package_quotas(all_packages=True) and\
                package not in self._get_package_to_users_map():
            self._save_package_limits(package, soft, hard)
            return
        # Check package existance
        try:
            self._get_package_to_users_map(package)
        except NoSuchPackageException:
            if sync:
                return
        # Example: {'/dev/sda1': ['502', '504', '515', '521', '501']}
        device_user_map = self._get_device_user_map()
        saved_quotas = self._get_current_quotas()
        for device in device_user_map.keys():
            for uid in self._get_package_to_users_map(package):
                if uid not in device_user_map[device]:
                    continue
                _user = clpwd.get_names(int(uid))[0]
                if IS_DA and is_quota_inheritance_enabled():
                    panel = clcontrollib.DirectAdmin()
                    # check the real user's package and save his quotas (instead of setting `DEFAULT` package ones)
                    # this is only DA's specific
                    _real_package = panel._get_user_package(_user)
                    if _real_package != package:
                        _real_quotas = self._get_package_quotas(_real_package, True)
                        soft = _real_quotas[_real_package]['inodes_soft']
                        hard = _real_quotas[_real_package]['inodes_hard']
                data = self._combine_package_limits(uid=uid, soft=soft, hard=hard)
                if not data:
                    continue
                if soft == '-1':
                    soft_limit = '0'
                else:
                    soft_limit = data[0] or '0'
                if hard == '-1':
                    hard_limit = '0'
                else:
                    hard_limit = data[1] or '0'
                try:
                    if not self.limits_are_equal((saved_quotas[uid]['inodes_soft'], saved_quotas[uid]['inodes_hard']),
                                                 (soft_limit, hard_limit)):
                        std_in.append(
                            '%s %s %s %s %s' % (
                                uid,
                                saved_quotas[uid]['bytes_soft'], saved_quotas[uid]['bytes_hard'],
                                soft_limit, hard_limit))
                except KeyError:
                    pass # skip error when qouta is on but not configured
            if len(std_in) == 0:
                continue
            std_in = ('\n'.join(std_in) + '\n')
            if only_store:
                if device not in self._device_quota:
                    self._device_quota[device]  = ''
                self._device_quota[device] = self._device_quota[device] + std_in
            else:
                cmd = [QuotaWrapper.SETQUOTA, '-bu', device]
                run_command(cmd, std_in=std_in)
            std_in = []
        if save:
            self._save_package_limits(package, soft, hard)

def synchronize(self):
        """
        Read limits from file and applies them to packages and users
        """
        self._check_admin()
        # Get all packages with limits
        package_limits = self._get_all_packages_with_limits(True)
        for package in package_limits.keys():
            soft, hard = package_limits[package]
            self.set_package_limit(package, soft, hard, False, only_store=True)
        # Clear internal quotas cache
        self._quota = {}
        # Set user's individual limits
        if self._dh.has_section('users'):
            for uid in self._dh.options('users'):
                try:
                    # Check user presence
                    self._fetch_homedir(uid)
                    limits = self._dh.get('users', uid).split(':')
                    if len(limits) < 2:
                        continue
                    soft, hard = limits
                    # save=False
                    self.set_user_limit(uid, soft, hard, False, only_store=True)
                except NoSuchUserException:
                    self._dh.remove_option('users', uid)
            self._write_data()
        for device in self._device_quota.keys():
            cmd = [QuotaWrapper.SETQUOTA, '-bu', device]
            run_command(cmd, std_in=self._device_quota[device])

def save_user_cache(self):
        """
        Caches the limits to non-privileged user to see them
        """
        self._check_admin()
        cache_content = []
        # get data from repquota utility (for root), else from /etc/container/cl-quotas.cache file
        current_quotas = self._get_current_quotas()
        for k in sorted(list(current_quotas.keys()), key=int):
            cache_content.append([k] + list(map((lambda x: current_quotas[k][x]), self._fields)))
        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 _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):
        '''
        Gets ConfigParser handler for future use
        '''
        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):
        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)

@staticmethod
    def limits_are_equal(limits1, limits2):
        """
        Compare tuples
        :param limits1: tuple
        :param limits2: tuple
        :return: True of tuple 1 equal to tuple 2
        """
        if limits1 == limits2:
            return True
        return False

def _apply_to_all_if_not_set(self, soft=None, hard=None):
        """
        Applies limits to all users if no other (user or package) ones has
        not been set for them
        """
        std_in = []
        device_user_map = self._get_device_user_map()
        saved_quotas = self._get_current_quotas()
        for device in device_user_map.keys():
            cmd = [QuotaWrapper.SETQUOTA, '-bu', device]
            for uid in self._get_list_of_uids():
                # if uid not placed on current device or not owned by default package, pass it
                if uid not in device_user_map[device] or DEFAULT_PACKAGE not in self._get_uid_to_packages_map(uid):
                    continue
                data = self._combine_default_limits(uid=uid, soft=soft, hard=hard)
                if not data:
                    continue
                if not self.limits_are_equal((saved_quotas[uid]['inodes_soft'], saved_quotas[uid]['inodes_hard']), data):
                    std_in.append('%s %s %s %s %s' % (
                        uid,
                        saved_quotas[uid]['bytes_soft'],
                        saved_quotas[uid]['bytes_hard'],
                        data[0] or '0',
                        data[1] or '0'))
            if len(std_in) == 0:
                continue
            std_in = ('\n'.join(std_in) + '\n')
            run_command(cmd, std_in=std_in)
            std_in = []

def _combine_user_limits(self, uid, soft=None, hard=None):
        """
        Determines user limits taking into account
        saved package and default ones
        """
        u_soft, u_hard = self._get_saved_user_limits_if_none(uid=uid, soft=soft, hard=hard)
        p_soft, p_hard = None, None
        for package in self._get_uid_to_packages_map(uid):
            # Get user's package limits
            p_soft, p_hard = self._get_saved_package_limits_if_none_or_unlim(package=package, soft=u_soft, hard=u_hard)
        # Set absent limits to uid=0 limits
        d_soft, d_hard = self._get_saved_user_limits_if_none_or_unlim(uid='0', soft=p_soft, hard=p_hard)
        if d_soft is None:
            d_soft = '0'
        if d_hard is None:
            d_hard = '0'
        return d_soft, d_hard

def _combine_package_limits(self, uid, soft=None, hard=None):
        """
        Determines package limits taking into account
        saved user and default ones
        """
        # Get current user limits
        t_soft, t_hard = self._get_saved_user_limits_if_none(uid=uid)
        # If both limits present, do nothing
        if t_soft and t_hard:
            return ()
        # Inherit package limits
        if t_soft is None:
            t_soft = soft
        if t_hard is None:
            t_hard = hard
        # If package limits absent, use default limits
        return self._get_saved_user_limits_if_none_or_unlim(uid='0', soft=t_soft, hard=t_hard)

def _combine_default_limits(self, uid, soft=None, hard=None):
        """
        Determines default limits taking into account saved user and package ones
        :param soft: soft limit from uid=0
        :param hard: hard limit from uid=0
        """
        # Get user's limits from clquota.dat
        t_soft, t_hard = self._get_saved_user_limits_if_none(uid=uid)
        if t_soft is not None:
            soft = t_soft
        if t_hard is not None:
            hard = t_hard
        if soft == '-1':
            soft = '0'
        if hard == '-1':
            hard = '0'
        return soft, hard

def _get_saved_user_limits_if_none(self, uid, soft=None, hard=None):
        """
        Retrives saved user limits if none has passed
        """
        try:
            user_soft, user_hard = self._dh.get('users', uid).split(':')
            if soft is None and user_soft != '0':
                soft = user_soft
            if hard is None and user_hard != '0':
                hard = user_hard
        except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
            pass
        soft = self._check_limit(soft)
        hard = self._check_limit(hard)
        return soft, hard

def _get_saved_user_limits_if_none_or_unlim(self, uid, soft=None, hard=None):
        """
        Applies saved user limits if none or unlimit has been passed
        """
        try:
            user_soft, user_hard = self._dh.get('users', uid).split(':')
            # Replace -1 to 0 for set unlimited limit
            if user_soft == '-1':
                user_soft = '0'
            if user_hard == '-1':
                user_hard = '0'
            if (soft is None or soft == '0') and user_soft != '0':
                soft = user_soft
            if (hard is None or hard == '0') and user_hard != '0':
                hard = user_hard
        except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
            pass
        soft = self._check_limit(soft)
        hard = self._check_limit(hard)
        return soft, hard

def _get_saved_package_limits_if_none(self, package, soft=None, hard=None):
        """
        Applies saved package limits if none has passed
        """
        if package != DEFAULT_PACKAGE:
            try:
                pack_soft, pack_hard = self._dh.get('packages', package).split(':')
                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_saved_package_limits_if_none_or_unlim(self, package, soft=None, hard=None):
        """
        Applies saved package limits if none or unlimit has been passed
        """
        if package != DEFAULT_PACKAGE:
            try:
                pack_soft, pack_hard = self._dh.get('packages', package).split(':')
                if (soft is None or soft == '0') and pack_soft != '0':
                    soft = pack_soft
                if (hard is None or hard == '0') 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_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 == 'default':
                    # Because "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):
        """
        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)
        self._uid_to_packages_map = defaultdict(list)

for uid, uid_data in users_packages.items():
            s_uid = str(uid)
            package = uid_data['package'] if uid_data['package'] is not None else 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 DEFAULT_PACKAGE, [])

packages_users.setdefault(DEFAULT_PACKAGE, [])

return packages_users

def _load_package_uids_data(self):
        """
        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 'default' package
        if not self._check_present_panel():
            packages[DEFAULT_PACKAGE] = list(map(str, _get_users_list()))
            self._uid_to_packages_map = dict((i, DEFAULT_PACKAGE) for i in packages[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 'default' package to list
        if DEFAULT_PACKAGE not in self._all_package_list:
            self._all_package_list.append(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(
                dict(
                    ((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 = {}
        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):
        """
        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'
        soft, hard = self._get_saved_user_limits_if_none(uid, soft, hard)
        if (soft is None or soft == '0') and\
                (hard is None or 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'
        soft, hard = self._get_saved_package_limits_if_none(package, soft, hard)
        if (soft is None or soft == '0') and (hard is None or 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'
        if self.limits_are_equal((old_cpanel_data.get('inodes_soft', '0'),
                                  old_cpanel_data.get('inodes_hard', '0')),
                                 (quotas_data['inodes_soft'], quotas_data['inodes_hard'])):
            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

System Information
*ReClick For Close
File : BlackDragon /about/function.php

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2023-09-24 06:29:09
..	dir	-	root/root	0755	2024-03-18 03:02:40
__pycache__	dir	-	root/root	0755	2023-09-24 06:30:01
__init__.py	text/plain	49.81 KB	root/root	0644	2023-09-12 09:06:12
cl_quota.py	text/plain	9.84 KB	root/root	0644	2023-09-12 09:06:12
utils.py	text/x-python	2.54 KB	root/root	0644	2023-09-12 09:06:12