FileMgr

✏️ /opt/cloudlinux/venv/lib/python3.11/site-packages/lvectllib.py

✕ Tutup

# coding=utf-8
# Liblve functions lib

#
# 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 contextlib
import copy
import errno
import json
import math
import mmap
import os
import pwd
import re
import subprocess
import sys
import syslog
import warnings
import xml.dom.minidom as xml
from builtins import range
from functools import partial
from mmap import PAGESIZE
from typing import Dict, List, Optional, Text, Tuple, TypedDict  # NOQA

import unshare

import clcommon
import cldetectlib
import lveapi
from clcommon.const import Feature
from clcommon.cpapi import admins, get_main_username_by_uid, is_panel_feature_supported, reseller_users
from clcommon.cpapi.cpapiexceptions import EncodingError
from clcommon.lock import acquire_lock
from clcontrollib import detect_panelclass
from clevents import reseller_limits_disabled_post, reseller_limits_enabled_post
from cllimits.lib import exec_utility
from cllvectl.log import get_subprocess_logger
from clveconfig.ve_config import BadVeConfigException, get_xml_config, save_xml
from clveconfig.ve_lock import LockFailedException, setup_global_lock
from lveapi import LVP_XML_TAG_NAME, Lve, NameMap, PyLve, PyLveError
from secureio import create_dir_secure, write_file_via_tempfile
from websiteisolation import config as ws_config
from websiteisolation import id_registry as ws_id_registry
from websiteisolation.exceptions import LvdError

GET_CP_PACKAGE_SCRIPT = '/usr/bin/getcontrolpaneluserspackages'

CPUINFO = '/proc/cpuinfo'
CORE_WEIGHT = 10000
DEFAULT_PACKAGE = "VE_DEFAULT"

NOIOPS = False

UMOUNT = '/bin/umount'
EXCLUDE_MOUNTS_CONF = '/etc/container/exclude_mounts.conf'

MULTI_FORMAT = 'multi'
SINGLE_FORMAT = 'single'

IS_DEBUG = int(os.environ.get('PYLVE_DEBUG', 0))

XML_PLESK_ID = 'plesk_id'   # XML attribute name to bind package name in the ve.cfg with plesk DB id

if not is_panel_feature_supported(Feature.LVE):
    pylve = None
    lve = None
else:
    pylve = PyLve(debug=IS_DEBUG)
    lve = Lve(py=pylve)

class LiblveSettings(TypedDict):
    ls_cpu: int | str
    ls_cpus: int
    ls_io: int
    ls_enters: int
    ls_memory_phy: int
    ls_nproc: int
    ls_iops: int

def create_liblve_settings(**kwargs) -> LiblveSettings:
    defaults: LiblveSettings = {
        'ls_cpu': 0,
        'ls_cpus': 0,
        'ls_io': 0,
        'ls_enters': 0,
        'ls_memory_phy': 0,
        'ls_nproc': 0,
        'ls_iops': 0
    }
    defaults.update(kwargs)
    return defaults

def lvp_list():
    """Raw kernel LVP list (/proc/lve/resellers), including per-user
    domain-isolation LVPs.  Reserved for isolation-aware internals; reseller-
    management code MUST iterate reseller_lvp_list() instead."""
    if lve.reseller_limit_supported():
        return lve.proc.lvp_id_list()
    return []

def _isolation_lvp_ids():
    """Per-user domain-isolation LVP ids (lvp<uid>).

The kernel registers an LVP named lvp<uid> in /proc/lve/resellers for every
    user that has per-domain isolation allowed/enabled (a marker in the
    websiteisolation config), even when the user has 0 registered domains.
    These are not resellers — they are maintained by the isolation feature —
    so reseller-management code must never treat them as reseller LVPs.

This is the single source of truth for "is this kernel LVP an isolation
    LVP"; it mirrors the same find_all_lve_ids_with_config() signal that
    lveapi uses to build the nested LVP/LVE hierarchy.  Note this is broader
    than _is_domain_isolated_user(), which additionally requires >= 1 domain:
    the kernel LVP exists on the marker alone (see CLOS-4513).

Disk-only: callable from non-lvectl entrypoints (cloudlinux-packages,
    clsummary) that do not initialize pylve, so this MUST NOT make a kernel
    call.  ws_config.find_all_lve_ids_with_config() reads /etc/container/lvd_ids
    and returns [] when the directory is absent, which covers both "kernel
    does not support domains" and "no isolation configured" without needing
    lve_kapi_ver().
    """
    return frozenset(ws_config.find_all_lve_ids_with_config())

def is_isolation_lvp(lvp_id):
    """True if lvp_id is a per-user domain-isolation LVP (see _isolation_lvp_ids)."""
    return lvp_id in _isolation_lvp_ids()

def reseller_lvp_list():
    """Kernel reseller LVPs, excluding per-user domain-isolation LVPs.

Use this everywhere reseller-management code enumerates resellers (destroy
    all, apply all, remove absent resellers, list-reseller, package limits,
    summaries).  Centralizing the phantom-isolation-LVP filter here means a new
    reseller command cannot reintroduce the CLOS-4370/4390/4419/4513 class of
    bug as long as it iterates this function instead of the raw lvp_list().

Backward compatible: on kernels without domain support the isolation set is
    empty, so the result equals lvp_list().
    """
    isolation = _isolation_lvp_ids()
    return [lvp_id for lvp_id in lvp_list() if lvp_id not in isolation]

def get_active_resellers():
    """
    Get list of resellers with activated reseller limits

:return: list of pairs (name, uid)
    """
    name_map = NameMap()
    name_map.link_xml_node()
    return name_map.load_from_node()

def is_active_reseller_limits(reseller_name):
    """
    Check whether giver reseller has activated reseller limits or not
    :return: bool
    """
    return reseller_name in (name for name, uid in get_active_resellers())

# TODO: py3 move it to cllib/long script
def raise_cpanel_encoding_error(e: EncodingError):
    """
    Since cPanel user can corrupt config file for some user with wrong encodings,
    we want to notify him that he should fix encoding problems with the link to documentation.
    Print error message and exit with code 1 or raise given exception if it isn't cPanel.
    :return: None
    """
    if not cldetectlib.is_cpanel():
        raise e

if JSON:
        json_format('multi',
                    ['ERROR', str(e)])
    else:
        print(e)
    sys.exit(1)

def get_global_lock(write=False):
    """
    ~~~~~~~~~~~~~~~~~~
    !!! DEPRECATED !!!
    ~~~~~~~~~~~~~~~~~~
    Please, use setup_global_lock instead if possible

Wrapper over setup_global_lock. If lock cannot be set,
    it will write message and close app

The only reason why it is here is legacy function
    check_result_and_exit that we use in TWO places
    :type write: bool
    :return: Nothing
    """
    try:
        setup_global_lock(write)
    except LockFailedException:
        check_result_and_exit(1, 'can`t get lock')

def check_result_and_exit(result, message):
    # on cl5 some func is unimplemented; so ENOSYS is not error;
    if result not in (0, -errno.ENOSYS):
        if JSON:
            json_format(MULTI_FORMAT, ['ERROR', f'lvectl: {message}'])
        else:
            print(f'lvectl: Error: {message}')
        sys.exit(result)

# Default parameters for lve
LVE_DEFAULT = {
    'cpu': 25,
    'ncpu': 1,
    'io': 25,
    'ep': 20,
    'mem': 0,
    'pmem': 262144,
    'nproc': 0,
    'iops': 1024

}
MEM_DEFAULT_CL5 = 262144
# Default parameters for lvp
LVP_DEFAULT = {
    'cpu': 100,
    'ncpu': 1,
    'io': 0,
    'ep': 0,
    'mem': 0,
    'pmem': 0,
    'nproc': 0,
    'iops': 0
}

LIMITS_LIST_NAME = ['ncpu', 'cpu', 'io', 'mem', 'pmem', 'nproc', 'iops', 'ep']

LVE_VERSION = 4
JSON = False
BYTES_FLAG = False

# defined structures for liblve and turples for functions
lve_settings = ''
setup_data = ''  # type: dict

# dict with user-packages relations
# keys = int UID or str package name
# data = string package name
packages_users = {}

# defined ve.cfg variables
ve_cfg = ''
ve_lveconfig = ''
ve_default = ''
ve_lve = ''
ve_lvp = ''  # for resellers limits
ve_defaults = ''  # type: dict
ve_package = ''
ubc = 'false'  # TODO: looks like not used anymore, check and remove it
ve_enter_by_name = ''
ve_binary = ''
ve_cfg_version = ''

# Set JSON if json output required
def set_json(json_flag):
    global JSON
    JSON = json_flag

def set_bytes(bytes_flag):
    global BYTES_FLAG
    BYTES_FLAG = bytes_flag

def get_fields():
    if NOIOPS and LVE_VERSION == 8:
        version = 'noiops_8'
    elif LVE_VERSION == 8:
        version = '8'
    elif LVE_VERSION == 6:
        version = '6'
    else:                       # LVE_VERSION == 4
        version = '4'
    fields = {
        'noiops_8': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO'],
        '8': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO','IOPS'],
        '6': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO'],
        '4': ['ID','SPEED','VMEM','EP','IO']
    }[version]
    if JSON:
        speed_idx = fields.index('SPEED') + 1
        return (fields[:speed_idx] + ["CPU"] + fields[speed_idx:])
    return fields

# Create structure
def init(lve_ver=None):
    global LVE_VERSION
    if lve_ver is None:
        lve_ver = clcommon.get_lve_version()
        if lve_ver[0] is None:
            raise RuntimeError('get_lve_version failed')
        LVE_VERSION = lve_ver[0]
    else:
        LVE_VERSION = lve_ver
    global lve_settings

lve_status = pylve.initialize()
    if not lve_status:
        raise RuntimeError('init_lve() failed.')

lve_settings = pylve.liblve_settings()

# we use /proc/cpuinfo to get cpu speed, but unfortunately
# it returns current CPU MHZ, which is different in lve environment
# and we cannot get right speed value there
# FIXME: LU-947
def _get_cpu_data_from_env():
    """Get cpu information from environment veriable"""
    packed_cpu_data = os.environ.get('CPU_DATA')
    if packed_cpu_data is None:
        return None

try:
        return json.loads(packed_cpu_data)
    except (TypeError, ValueError) as e:
        print('Invalid environment variable \'CPU_DATA\' format', str(e))
        sys.exit(1)

def get_cpu_data():
    """
    Parse /proc/cpuinfo

return [NumProc, frequency in MHZ]

"""

cpuinfo = {}
    procinfo = {}
    nprocs = 0
    try:
        # f = open(CPUINFO, 'r')
        with open(CPUINFO, 'r', encoding='utf-8') as f:
            for line in f:
                if not line.strip():
                    # end of one processor
                    cpuinfo[f'proc{nprocs}'] = procinfo
                    nprocs = nprocs + 1
                    # Reset
                    procinfo = {}
                else:
                    if len(line.split(':')) == 2:
                        procinfo[line.split(':')[0].strip()] = line.split(':')[1].strip()
                    else:
                        procinfo[line.split(':')[0].strip()] = ''
    except IOError:
        print(f'lvectl: Error: Can`t open {CPUINFO}.')
        sys.exit(1)

return [nprocs, cpuinfo['proc0']['cpu MHz']]

# It's extremely rare case when CPU changes at runtime so we will use cached
CPUINFO_DATA = _get_cpu_data_from_env() or get_cpu_data()

def convert_from_old_cpu(data, lncpu=0):
    """
    Try converting to kernel format from old CPU format (percentage of whole
    cpu) and optionally the NCPU format. Return whichever is less.

:param data: string presumably in old CPU format
    :param lncpu: integer number of cores limit
    """
    data = str(data)
    lncpu = lncpu or 0
    cpu_data = CPUINFO_DATA
    ncpu = int(cpu_data[0])
    cpu_percent = re.match(r'\d{1,2}0?$', data)  # 0-100
    if cpu_percent is not None:
        data = int(data)
        if 0 < data <= 100:
            from_cpu_limit = int(round(CORE_WEIGHT // 100 * ncpu * data))
            if lncpu == 0:
                return from_cpu_limit
            return min(lncpu * CORE_WEIGHT, from_cpu_limit)
    return None

def convert_from_speed_percent(data):
    """
    Try converting cpu limit from SPEED in percentage of one CORE format to
    kernel format.
    """
    data = str(data)
    cpu_data = CPUINFO_DATA
    ncpu = int(cpu_data[0])
    percent = re.match(r'\d+(?:\.\d+)?%$', data)  # *%
    if percent is not None:
        percent = float(data.replace('%', ''))
        if percent > ncpu * 100:
            percent = ncpu * 100
        if percent > 0:
            return int(round(CORE_WEIGHT // 100 * percent))
        return None
    return None

def convert_from_speed_hz(data):
    """
    Try converting cpu limit from SPEED in mhz/gzh format to kernel format.
    """
    data = str(data)
    cpu_data = CPUINFO_DATA
    ncpu = int(cpu_data[0])
    cpu_freq = float(cpu_data[1])
    pattern = re.compile(r'(?P<freq>\d+(?:\.\d+)?)(?P<suffix>mhz|ghz)+$', re.IGNORECASE)
    match = pattern.match(data)  # *mhz\ghz
    if match is not None:
        suffix = match.group('suffix')
        freq = float(match.group('freq'))
        if suffix.upper() == 'GHZ':
            freq = freq * 1000
        if freq > cpu_freq * ncpu:
            freq = cpu_freq * ncpu
        if freq > 0:
            return int(round(freq * CORE_WEIGHT / cpu_freq))
    return None

def convert_from_speed(data):
    """
    Try converting cpu limit value from either SPEED limit format
    (percentage of CORE or mhz/ghz) to kernel format.
    """
    return (
        convert_from_speed_percent(data)
        or convert_from_speed_hz(data)
    )

def convert_to_kernel_format(data, lncpu=0):
    """
    Convert different variants of cpu limit to kmod ver 8 variant

:param data: Value in old CPU format or SPEED with % or mhz/ghz.
    :param lncpu: Limit in old NCPU format.
    :return: CPU limit in kmod ver 8+ format or None for bad format
    """
    from_cpu = convert_from_old_cpu(data, lncpu)
    if from_cpu is not None:
        return from_cpu
    from_speed_percent = convert_from_speed_percent(data)
    if from_speed_percent is not None:
        return from_speed_percent
    from_speed_hz = convert_from_speed_hz(data)
    if from_speed_hz is not None:
        return from_speed_hz

return None

def speed_to_old_cpu(speed):
    """
    convert speed to old cpu format
    args: cpu limit in speed value
    return: old cpu limit format
    """
    cpu_data = CPUINFO_DATA
    nproc = int(cpu_data[0])
    speed = str(speed)
    if '*' in speed:
        return '*' + str(int(round(int(speed.lstrip('*')) // nproc)))
    return str(int(round(int(speed) // nproc)))

@contextlib.contextmanager
def temporary_lve(settings):
    # type: (pylve.liblve_settings) -> contextlib.GeneratorContextManager
    """
    Run subprocess in lve with pseudo-random id and given limits
    """
    pylve.initialize()
    lve_id = pylve.get_available_lve_id()
    try:
        pylve.lve_setup(lve_id, settings)
    except PyLveError:
        syslog.syslog(syslog.LOG_ALERT, f"Unable to setup lve with id {lve_id}, "
                                        "something is wrong, check dmesg for details")
        raise

try:
        yield lve_id
    finally:
        pylve.lve_destroy(lve_id)

def make_liblve_settings(settings: LiblveSettings):
    # type: (LiblveSettings) -> pylve.liblve_settings
    """
    Just a nice user-friendly constructor of liblve_settings object
    You can pass the following ls_cpu and ls_cpus values:
    - in percents of one core (just ls_cpu='75%', ls_cpus will be ignored)
    - in old 'CPU' format (two arguments, ls_cpu and ls_cpus required, both int)
    """
    s = pylve.liblve_settings()
    s.ls_cpu = convert_to_kernel_format(settings['ls_cpu'], lncpu=settings['ls_cpus'])
    s.ls_io = settings['ls_io']
    s.ls_enters = settings['ls_enters']
    s.ls_nproc = settings['ls_nproc']
    s.ls_iops = settings['ls_iops']
    # convert memory from bytes to mempages
    s.ls_memory_phy = int(math.ceil(1. * settings['ls_memory_phy'] / PAGESIZE))
    return s

def get_ve_lve_user_uid(ve_lve_element):
    user_uid = str(ve_lve_element.getAttribute('id'))
    if not user_uid:
        user_name = ve_lve_element.getAttribute('user')
        user_uid = pwd.getpwnam(user_name).pw_uid
    return int(user_uid)

def json_format(error_type, data, extensions=None):
    """
    Print output in json as:
    {"status": "ERROR/OK", "msg": "Some Message", "ext1": "foo", "ext2": "bar"}
    where "status" and "msg" field are mandatory

:param str error_type: Either MULTI_ERROR or SINGLE_ERROR
    :param list data: List with a status string and a message string
    :param dict extensions: Some additional fields for the final json object
    :return: None
    """
    result = {'status': str(data[0])}
    if error_type == MULTI_FORMAT:
        result['msg'] = str(data[1])
    if extensions is not None:
        result.update(extensions)
    print(json.dumps(result))

def check_def_value(xml, ve_defaults, ve_cfg, val, default):
    try:
        ve_defaults[val] = int(ve_default.getElementsByTagName(val)[0].getAttribute('limit'))
    except (ValueError, IndexError, TypeError):
        ve_defaults[val] = default[val]
        node = ve_cfg.createElement(val)
        node.setAttribute('limit',str(default[val]))
        try:
            xml.appendChild(node)
        except Exception:
            pass

def xml_filter_tag(node, tag):
    return [_ for _ in node.childNodes if isinstance(_, xml.Element) and _.tagName == tag]

def xml_filter_first(node, tag, attr=None, attr_val=None):
    for child_node in xml_filter_tag(node, tag):
        if attr is not None and not child_node.hasAttribute(attr):
            continue
        if attr_val is not None and child_node.getAttribute(attr) != attr_val:
            continue
        return child_node

def get_child_tag_atrr(node, tag, attr):
    filtered_child_node = xml_filter_first(node=node, tag=tag, attr=attr)
    if filtered_child_node is None:
        raise IndexError()
    return filtered_child_node.getAttribute(attr)

def set_child_tag_atrr(node, tag, attr, val):
    """
    Find in children nodes node with tag and setup attribute
    insted el.getElementsByTagName not search recursiveli in tree
    """
    first_child_node = xml_filter_tag(node, tag)[0]
    first_child_node.setAttribute(attr, str(val))

def _load_config_wrapper():
    """Load config from ve.cfg"""
    global ve_cfg
    global ve_lveconfig

try:
        ve_cfg, ve_lveconfig = get_xml_config()
    except BadVeConfigException as e:
        if JSON:
            json_format(MULTI_FORMAT, ['ERROR', str(e)])
        else:
            print(str(e))
        sys.exit(1)

def _load_default_limits(lvp_id: int, lvp_defaults: bool):
    """Load default limits

:param int lvp_id: lvp id
    :param bool lvp_defaults: load reseller's default limits instead of global
    :return: dict with default limits
    """
    global ve_default
    global ubc

ubc = 'true'

try:
        if not lvp_id:
            ve_default = xml_filter_tag(ve_lveconfig, 'defaults')[0]
            return LVE_DEFAULT

if lvp_defaults:
            ve_default = ve_cfg.createElement('defaults')
            return LVP_DEFAULT

defaults_root_node = xml_filter_first(ve_lveconfig, LVP_XML_TAG_NAME, 'id', str(lvp_id))
        if defaults_root_node:  # if no such reseller with lvp in config
            ve_default = defaults_root_node.getElementsByTagName('defaults')[0]
        elif hasattr(ve_default, 'cloneNode'):
            ve_default = ve_default.cloneNode(ve_default)
        else:
            # ve_default not yet initialised — fall back to global defaults
            ve_default = xml_filter_tag(ve_lveconfig, 'defaults')[0]
        return LVE_DEFAULT

except IndexError:
        if JSON:
            json_format('multi', ['WARNING', 'default section error in ve.cfg'])
            sys.exit(1)
        else:
            print('warning: default section error in ve.cfg')

return LVE_DEFAULT

def _all_config_elements_loaded():
    return all(x != '' for x in (ve_lve, ve_lvp, ve_package, ve_binary, ve_enter_by_name, ve_cfg_version))

def _load_config_elements():
    """Load all config elements from ve.cfg"""
    global ve_lve
    global ve_lvp
    global ve_package
    global ve_binary
    global ve_enter_by_name
    global ve_cfg_version

ve_lve = ve_lveconfig.getElementsByTagName("lve")
    ve_package = ve_lveconfig.getElementsByTagName("package")
    ve_lvp = ve_lveconfig.getElementsByTagName(LVP_XML_TAG_NAME)
    lve.map.name_map.link_xml_node(ve_lveconfig)

enter_by_name_elems = ve_lveconfig.getElementsByTagName('enter-by-name')
    if len(enter_by_name_elems) > 0:
        ve_enter_by_name = enter_by_name_elems[0]
    else:
        ve_enter_by_name = ve_cfg.createElement('enter-by-name')
        ve_lveconfig.appendChild(ve_enter_by_name)

ve_binary = ve_enter_by_name.getElementsByTagName('binary')

# expected version tag in next format
    #
    # <lveconfig>
    #   <version>2</version>
    #   <system>
    #   ....
    #   </system>
    #   ....
    # </lveconfig>

cfg_version_elems = ve_lveconfig.getElementsByTagName('version')
    ve_cfg_version = int(cfg_version_elems[0].firstChild.nodeValue) if len(cfg_version_elems) > 0 else 1

def _load_ve_defaults(default_limits: Dict[str, int]):
    """Create ve_defaults dict with default values for all limits

:param dict default_limits: default limits for lve or lvp
    """
    global ve_cfg
    global ve_defaults
    global ve_default

ve_defaults = {}
    check_def_val = partial(
        check_def_value,
        xml=ve_default,
        ve_defaults=ve_defaults,
        ve_cfg=ve_cfg,
        default=default_limits,
    )
    check_def_val(val='ncpu')

try:
        speed = ve_default.getElementsByTagName('cpu')[0].getAttribute('limit')
        ve_defaults['cpu'] = convert_to_kernel_format(speed, lncpu=ve_defaults['ncpu'])
    except (ValueError, IndexError, TypeError):
        ve_defaults['cpu'] = convert_to_kernel_format(default_limits['cpu'], lncpu=ve_defaults['ncpu'])
        cpu = ve_cfg.createElement('cpu')
        cpu.setAttribute('limit', str(default_limits['cpu']))
        try:
            ve_default.appendChild(cpu)
        except Exception:
            pass

try:
        ve_defaults['ep'] = int(ve_default.getElementsByTagName('other')[0].getAttribute('maxentryprocs'))
    except (ValueError, IndexError, TypeError):
        ve_defaults['ep'] = default_limits['ep']
        ep = ve_cfg.createElement('other')
        ep.setAttribute('maxentryprocs', str(default_limits['ep']))
        try:
            ve_default.appendChild(ep)
        except Exception:
            pass

check_def_val(val='io')
    if LVE_VERSION > 5:
        check_def_val(val='mem')
    else:
        check_def_val(val='mem', default={'mem': MEM_DEFAULT_CL5})  # pylint: disable=redundant-keyword-arg
    check_def_val(val='pmem')
    check_def_val(val='nproc')
    check_def_val(val='iops')

_check_defaults_for_nones()

def _check_defaults_for_nones():
    """Check that all default values are not None"""
    global ve_defaults

for key, value in ve_defaults.items():
        if value is not None:
            continue

err_msg = f'ERROR: Incorrect {key} default value'
        if JSON:
            json_format('multi', ['ERROR', err_msg])
        else:
            sys.stderr.write(f'{err_msg}\n')
        sys.exit(1)

def get_XML_cfg(lvp_id=0, lvp_defaults=False, load_config_elements=True):
    """
    :param bool lvp_defaults: load reseller's default limits instead of global
    :param int lvp_id: lvp id to load customise defaults
    """

_load_config_wrapper()
    default_limits = _load_default_limits(lvp_id, lvp_defaults)

# load config elements if load_config_elements=True or they were not loaded before
    if load_config_elements or not _all_config_elements_loaded():
        _load_config_elements()

_load_ve_defaults(default_limits)

def check_value(val, el, ve_defaults, setup_data):
    try:
        value = int(get_child_tag_atrr(el, tag=val, attr='limit'))
        setup_data[val] = value
        return value
    except (ValueError, IndexError, TypeError):
        return int(ve_defaults[val])

def _load_resellers_xml_data(reseller, xml_config_load_elements=True):
    """
    This function is a pure workaround for our ugly globals-based API which
    should be fixed partially with LU-496, because there is no clean way
    to retrieve reseller's data from ve.cfg without touching globals
    :param reseller: reseller name
    :return: Nothing. It just updates some globals
    """
    # TODO after LU-496 we should read and cache all reseller's settings
    name_map = lveapi.NameMap()
    name_map.link_xml_node()
    reseller_id = name_map.get_id(reseller)
    get_XML_cfg(reseller_id, load_config_elements=xml_config_load_elements)

def prepare_setup_data(plan_id=None, reseller=None, lve_id=None):
    # type: (Optional[Text], Optional[Text]) -> None
    """
    Put limit values that will be applied later in a global variable `setup_ve`.

:param plan_id: package
    :param reseller:
        If reseller is None we only inherit from admin packages.
        In that case we ignore all tags in ve.cfg with a "reseller" attribute.
    """

global setup_data
    setup_data = copy.copy(ve_defaults)
    if plan_id is not None:
        res_pkg_dict = get_reseller_packages_map()
        # LU-510: Investigate the problem with reseller's list, part 2.
        #   Fix applying limit for reseller user with admin package
        if reseller is not None and reseller in res_pkg_dict and plan_id in res_pkg_dict[reseller]:

def is_needed_plan(el):
                return el.getAttribute('id') == plan_id and el.getAttribute('reseller') == reseller

else:
            if cldetectlib.is_da() and lve_id is not None:
                try:
                    user_pwd = pwd.getpwuid(lve_id)
                    filename = f'/usr/local/directadmin/data/users/{user_pwd.pw_name}/user.conf'
                    with open(filename, encoding='utf-8') as f:
                        text = f.read()
                except Exception:
                    text = ''
                # LU-1663 --> LU-3410:
                #   it is normal situation for DA to set DA(not LVE) user package to "custom"
                #   in this case the package is still correct and this warning should not be logged
                if 'package=custom' not in text:
                    syslog.syslog(
                        syslog.LOG_ALERT,
                        f"Package for user with id {lve_id} is incorrect, please recover it using Note from "
                        "https://docs.cloudlinux.com/cloudlinux_os_components/#installation-enabling-and-disabling",
                    )

# Ignore all tags in ve.cfg with a "reseller" attribute.
            def is_needed_plan(el):
                return el.getAttribute('id') == plan_id and not el.getAttribute('reseller')

# Example command when `ve_package` is not empty:
        #   cloudlinux-packages set --json --for-reseller root --package kekage --pmem 994 --nproc 77
        # `ve_package` is set by reading ve.cfg in `get_XML_cfg`.
        # <package> tag is added to ve.cfg in `package_set_ext`.
        for el in ve_package:
            if is_needed_plan(el):
                lncpu = check_value('ncpu', el, ve_defaults, setup_data)
                try:
                    cpu = int(convert_to_kernel_format(get_child_tag_atrr(el, tag='cpu', attr='limit'), lncpu=lncpu))
                    setup_data['cpu'] = cpu
                except (ValueError, IndexError, TypeError):
                    pass
                check_value('io', el, ve_defaults, setup_data)
                if (ubc == 'true'):
                    check_value('mem', el, ve_defaults, setup_data)
                else:
                    setup_data['mem'] = 0
                try:
                    ep = int(get_child_tag_atrr(el, tag='other', attr='maxentryprocs'))
                    setup_data['ep'] = ep
                except (ValueError, IndexError, TypeError):
                    pass
                check_value('nproc', el, ve_defaults, setup_data)
                check_value('pmem', el, ve_defaults, setup_data)
                check_value('iops', el, ve_defaults, setup_data)

def umount_dir(path):
    try:
        # run the "umount" command and suppress it's output, return True when child exit code is not zero
        with subprocess.Popen(
            [UMOUNT, "-l", path],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        ) as proc:
            proc.communicate()
            return proc.returncode != 0
    except OSError:
        check_result_and_exit(-1, f'failed to run "{UMOUNT} -l {path}"')

def prepare_mounts():
    """
    Unmount all paths from /proc/mounts that match regular expressions from /etc/container/exclude_mounts.conf file
    """
    if not os.path.isfile(EXCLUDE_MOUNTS_CONF):
        return
    reg_exp_list = []
    try:
        with open(EXCLUDE_MOUNTS_CONF, 'r', encoding='utf-8') as conf:
            for r in conf:
                pattern = r.strip()
                if pattern:
                    reg_exp_list.append(re.compile(pattern))
    except IOError:
        check_result_and_exit(-1, f'failed to read {EXCLUDE_MOUNTS_CONF}')
    if not reg_exp_list:
        return
    unshare.unshare(unshare.CLONE_NEWNS)
    try:
        with open('/proc/mounts', 'r', encoding='utf-8') as f:
            mounts = [m.split()[1] for m in f.readlines()]
    except (IndexError, IOError):
        check_result_and_exit(-1, 'failed to parse /proc/mounts')
    ATTEMPTS = 10
    for _ in range(ATTEMPTS):
        error = False
        for mount in mounts:
            for reg_exp in reg_exp_list:
                m = reg_exp.search(mount)
                if m:
                    error = umount_dir(mount) or error
                    break
        if not error:
            break

def lve_start():
    """
    Start LVE engine and initialize default mount namespace for LVE
    """
    MOUNT_CMD = '/bin/mount --make-rprivate / >/dev/null 2>&1'
    try:
        subprocess.call(MOUNT_CMD, shell=True, executable='/bin/bash')
    except OSError:
        print('Error: failed to execute', MOUNT_CMD)
    prepare_mounts()
    pylve.lve_start(err_msg='Can`t init lve default settings')

def lve_create(lve_id, ignore_error=False):
    """
    Create LVE container for given ID
    :type lve_id: int
    :type ignore_error: bool
    :return: Nothing
    """
    pylve.lve_create(lve_id,
                     err_msg=f'lvectl: Can`t create lve with id {lve_id}; error code {{code}}',
                     ignore_error=ignore_error)

def lvp_create(lvp_id, ignore_error=False):
    """
    Create LVP container for given ID
    :type lvp_id: int
    :type ignore_error: bool
    :return: Nothing
    """
    pylve.lve_lvp_create(lvp_id,
                         err_msg=f'lvectl: Can`t create lvp with id {lvp_id}; error code {{code}}',
                         ignore_error=ignore_error)

def destroy_lvp_all():
    logger = get_subprocess_logger('lvectllib')

destroyed_list = []
    # Destroy every LVP, including per-user domain-isolation LVPs (lvp<uid>) —
    # they are rebuilt by the subsequent `apply all`.
    #
    # CLOS-4513: lvp_list() reads /proc/lve/resellers, which lists every LVP
    # flat, but in the cgroup hierarchy an isolation LVP is nested inside its
    # owner's reseller LVP (lvp<reseller>/lvp<uid>).  Destroying the parent
    # cascades to the nested child, so a flat-list entry captured here can be
    # gone by the time the loop reaches it; destroying it would return -ESRCH
    # and abort the whole loop (breaking cagefs.service ExecStartPre).  Re-check
    # existence and skip the already-removed entries — that is expected, not an
    # error.  Genuine destroy failures still propagate.
    for lvp_id in lvp_list():
        if not lve.proc.exist_lvp(lvp_id=lvp_id):
            logger.debug('destroy_lvp_all: lvp %s n(cascaded with its parent), skipping', lvp_id)
            continue
        logger.debug('destroy_lvp_all: destroying LVP with id %s', lvp_id)
        pylve.lve_lvp_destroy(lvp_id,
                              err_msg=f'lvectl: Can`t destroy lvp with id {lvp_id}; error code {{code}}')
        destroyed_list.append(lvp_id)
    return destroyed_list

def lvp_destroy(lvp_id):
    logger = get_subprocess_logger('lvectllib')

if lvp_id == 'all':
        destroy_lvp_all()
        return

logger.debug('lvp_destroy: destroying LVP with id %s', lvp_id)
    pylve.lve_lvp_destroy(lvp_id,
                          err_msg=f'lvectl: Can`t destroy lvp with id {lvp_id}; error code {{code}}')

# Destroy LVE container for ID
def lve_destroy(lve_id):
    logger = get_subprocess_logger('lvectllib')

cant_remove_msg = f'Can\'t remove lve {lve_id} from kernel - error code -3'
    cant_destroy_msg = f'Can`t destroy lve with id {lve_id}; error code {{code}}'
    destroyed = False

if lve_id == 'all':
        if lve.reseller_limit_supported():
            destroyed = bool(destroy_lvp_all())  # destroy all top level containers
        if len(list(lve.proc.lve_id_list())) > 0:
            for id_ in lve.proc.lve_id_list():
                logger.debug('lve_destroy all: destroying LVE with id %s', id_)
                lve.lve_destroy(id_, err_msg=cant_destroy_msg)
                destroyed = True
        else:
            destroyed = True  # empty list - all lve already destroyed
    else:
        if lve.proc.check_inside_list(lve_id) or \
                (lve.proc.resellers_supported() and lve.proc.detect_inside_lvp(lve_id) is not None):
            logger.debug('lve_destroy: destroying LVE with id %s', lve_id)
            lve.lve_destroy(lve_id, err_msg=cant_destroy_msg)
            destroyed = True
        elif not lve.proc.check_inside_list(lve_id):
            destroyed = True  # lve_id doesn`t exist so it`s already destroy!

if destroyed:
        if JSON:
            json_format(SINGLE_FORMAT, ['OK'])
    else:
        if JSON:
            json_format(MULTI_FORMAT, ['WARN', cant_remove_msg])
        else:
            print(f'warning: {cant_remove_msg}')

# Setup LVE for ID
def lve_setup(lve_id, lvp_id=0):
    if lvp_id and not lve.proc.exist_lvp(lvp_id):
        # create lve top container if not exist
        lvp_create(lvp_id)

with warnings.catch_warnings():  # convert all warning to exceptions
        warnings.filterwarnings('error')
        try:
            lve_settings.ls_io = int(setup_data['io'])
            lve_settings.ls_cpu = int(setup_data['cpu'])
            lve_settings.ls_cpus = int(setup_data['ncpu'])
            lve_settings.ls_memory = int(setup_data['mem'])
            lve_settings.ls_enters = int(setup_data['ep'])

if LVE_VERSION > 5:
                lve_settings.ls_memory_phy = int(setup_data['pmem'])
                lve_settings.ls_nproc = int(setup_data['nproc'])
            if LVE_VERSION > 6:
                lve_settings.ls_iops = int(setup_data['iops'])
            if lvp_id:
                if lve_id == 0:
                    pylve.lve_set_default(
                        lvp_id, lve_settings, err_msg=f'Can`t setup default settings for LVP {lvp_id}')
                else:
                    pylve.lve_lvp_setup(
                        lvp_id, lve_settings, err_msg=f'Can`t setup lvp with id {lvp_id}; error code {{code}}')
            elif lve_id == 0:
                pylve.lve_set_default(lve_settings, err_msg='Can`t setup default settings')
            else:
                pylve.lve_setup(
                    lve_id, lve_settings, err_msg=f'Can`t setup lve with id {lve_id}; error code {{code}}')

except RuntimeWarning as rw:
            # exit if caught a warning - can`t set limits to lve
            check_result_and_exit(1,'Can`t setup lve ' + str(lve_id) + '. RuntimeWarning excepted: ' + str(rw))

def get_package_and_reseller_by_lve_id(lve_id):
    """
    Get pair of package, reseller for lve_id
    :param lve_id: lve_id, UID with package, reseller
    :return: tuple of (package, reseller); Both can be None
    """
    global packages_users
    package = None
    reseller = None
    # hate this! backup global ref to packages_users
    old_packages_users = packages_users
    GetControlPanelUsers('list-users')
    # now packages_users should be dict of dicts:
    # {lve_id : {'package' : , 'reseller':}}
    temp_package = packages_users
    # restore global ref to packages_users
    packages_users = old_packages_users
    if isinstance(temp_package, dict):
        try:
            reseller = temp_package[lve_id]['reseller']
            # if reseller is empty then reseller is root/admin. return to None
            if not reseller:
                reseller = None
        except KeyError:
            reseller = None
        try:
            package = temp_package[lve_id]['package']
            # if package is empty  - return to undefined state
            if not package:
                package = None
        except KeyError:
            package = None

return package, reseller

# (rprilipskii): Honestly, relying on global variables for storage like this leaves a bad
# taste in my mouth, but I'd have to rewrite a more lvectl code than I'd prefer if I
# wanted to do it in a more pythonic way. Technical debt is a problem.
CACHED_EFFECTIVE_LIMITS = {}
EFFECTIVE_CACHE_FILE = "/var/run/cloudlinux/effective-normal-limits"
BURSTABLE_LIMITS_FLAG_FILE = "/opt/cloudlinux/flags/enabled-flags.d/burstable-limits.flag"

def write_effective_cache(reset=False):
    """
    Save calculated effective normal limits for an LVE to a cache file.

The cache file is stored in /var/run (tmpfs) for faster operations.
    It disappears on reboot, but lvectl apply all runs in the lvectl systemd service anyway,
    so the file will always reappear if the service functions normally.

If additional speed is desired, consider replacing the standard JSON lib
    with a faster one like rapidjson or orjson.

:param reset: If True, recreate the cache file from scratch with data available in the dict,
    instead of updating it, defaults to False
    :type reset: bool, optional
    """
    # NOTE: When running `lvectl apply all`, this write will happen at the very end
    # of the operation. Applying the limits, however, happens before this - right after
    # they're calculated.
    # This means a substantial time gap and a potential race condition if someone calls
    # `lvectl set` while `apply all` is not yet finished (e.g. cache and kernel having different limits).
    # Could be solved by makin limit application happen after all effective limits are
    # done calculating, and wrapping limit application and cache writing in the same filelock.

# Do nothing if the corresponding feature flag is not set
    if not os.path.exists(BURSTABLE_LIMITS_FLAG_FILE):
        return

# If there's no /var/run/cloudlinux directory, create it.
    try:
        effective_dir = os.path.dirname(EFFECTIVE_CACHE_FILE)
        if not os.path.isdir(effective_dir):
            var_run_dir = os.path.dirname(effective_dir)
            create_dir_secure(effective_dir, 0o600, 0, 0, var_run_dir)
    except OSError as e:
        print(f"Error: failed to create the folder {effective_dir}: {e}")
        raise

# Load already cached effective limits and merge them with
    # the limits calculated during the current lvectl run.
    # Unless we want to discard them and start over, that is.
    effective_cache_lock = acquire_lock(f"{EFFECTIVE_CACHE_FILE}.lock")
    with effective_cache_lock:
        if reset:
            effective_limits = {}
        else:
            try:
                if os.path.isfile(EFFECTIVE_CACHE_FILE):
                    with open(EFFECTIVE_CACHE_FILE, "r", encoding="utf8") as readfile:
                        effective_limits = json.load(readfile)
                else:
                    effective_limits = {}
            except json.JSONDecodeError as e:
                print(f"Error: failed to parse the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}")
                raise
            except OSError as e:
                print(f"Error: failed to read the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}")
                raise
        effective_limits.update(CACHED_EFFECTIVE_LIMITS)

try:
            write_file_via_tempfile(json.dumps(effective_limits), EFFECTIVE_CACHE_FILE, 0o600)
        except OSError as e:
            print(f"Error: failed to save the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}")

def cache_effective_limits(lve_id):
    """
    Cache the calculated effective normal limits for an LVE to a dictionary.

setup_data is a global dict that contains the LVE configuration last applied.
    Holds only one entry - during operations like lvectl apply all, we enter this
    function repeatedly with setup_data containing info for one processed LVE each time.

These limits can be used by the Burstable Limits components
    to set/reset burst or normal limits without having to go
    through effective limit calculation or invoking lvectl as a middleman.
    Comparing the cached limits to current active limits (in kernel)
    also shows whether or not the LVE has burst limits active at the moment.

This function gets called in lve_apply, which means it also runs within:
    * lvectl apply all
    * lvectl apply-many
    * lvectl set
    * and others, that also call lve_apply

:param lve_id: LVE ID for which the limits are being saved.
    :type lve_id: int
    """
    global CACHED_EFFECTIVE_LIMITS
    CACHED_EFFECTIVE_LIMITS[str(lve_id)] = setup_data

def _is_domain_isolated_user(lve_id: int) -> bool:
    """Return True if the user has at least one isolated domain registered.

CLOS-4370: the per-domain branch in lve_apply only does meaningful work
    when the user has at least one docroot in the lvd-map. A bare marker
    file from ``lvectl allow-domain-limits`` (with no ``enable-domain-limits``
    follow-up) or a 0-row hashtable left behind by ``lvd_map_remove`` after
    the last ``disable-domain-limits`` both leave the per-uid file in place,
    so existence alone is not enough — we must also check that the
    hashtable actually has entries.
    """
    if lve_id == 0 or not lve.py.domains_supported():
        return False
    if lve_id not in set(ws_config.find_all_lve_ids_with_config()):
        return False
    return bool(ws_id_registry.get_all_entries(lve_id))

def _get_all_domain_lve_ids() -> set:
    """Collect all known domain LVE IDs from the on-disk registries.

Domain LVEs (created by lvdctl set) appear in /proc/lve/list at root
    level, but lve_apply() must not process them as regular users — their
    limits are managed by apply_domain_lve_limits() inside the
    domain-isolated-user branch.
    """
    if not lve.py.domains_supported():
        return set()
    return ws_id_registry.get_all_domain_ids()

class DomainLveIdError(Exception):
    """Raised when lvectl set/delete/destroy targets a domain LVE ID."""

def _reject_if_domain_lve_id(lve_id: int) -> None:
    """Refuse to operate on domain LVE IDs from lvectl set/delete/destroy.

Domain LVEs are managed exclusively by lvdctl; lvectl operations
    on them corrupt the domain configuration or create phantom entries
    in ve.cfg.

The check only activates when at least one domain LVE is registered,
    so existing customers without domain isolation see no change in
    behaviour.
    """
    if lve_id <= 0 or not lve.py.domains_supported():
        return
    domain_ids = _get_all_domain_lve_ids()
    if not domain_ids:
        return
    min_domain_id = ws_id_registry.min_id()
    if lve_id >= min_domain_id or lve_id in domain_ids:
        raise DomainLveIdError(
            f"LVE ID {lve_id} is in the domain LVE range (>= {min_domain_id}) "
            f"or is a registered domain LVE. "
            f"Domain limits must be managed via lvdctl, not lvectl."
        )

# pylint: disable-msg=too-many-arguments
# Apply user settings
def lve_apply(lve_id, plan_id=None, result=False, reseller=None, out_node=None, lvp_id=0):
    """
    Aplly limits to LVE lve_id
    :param lve_id: lve id
    :type lve_id: int
    :param plan_id: package for user with lve_id. deprecated
    :type plan_id: string
    :param result: if True = don't apply limits. only create setup_data with actual limits
    :type result: boolean
    :param reseller: if True = plan_id is resellers plan. deprecated
    :type reseller: boolean
    :param out_node: node with limits for lve_id
    :type out_node: xml_node
    :param lvp_id: reseller container id; host container if 0
    """
    global setup_data
    global packages_users

old_packages_users = packages_users
    if lve_id != 0:
        GetControlPanelUsers('userid', lve_id)
        new_packages_users, packages_users = packages_users, old_packages_users

if ve_cfg == '':
        get_XML_cfg(lvp_id=lvp_id)

el = None
    if out_node is not None:
        el = out_node
    else:
        node_list = ve_lvp if lvp_id else ve_lve
        el = next(
            filter(lambda node: get_ve_lve_user_uid(ve_lve_element=node) == lve_id, node_list),
            None
        )

# reseller (lvp_id!=0) should not use package and default limits
    if lvp_id == 0:
        try:
            plan_id = new_packages_users[lve_id]['package']
        except (NameError, KeyError):
            plan_id = None

try:
            reseller = new_packages_users[lve_id]['reseller']
        except (NameError, KeyError):
            reseller = None

if el is not None:
        # get limits from package
        prepare_setup_data(plan_id, reseller=reseller)

# prepare custom limits for lve
        lncpu = check_value('ncpu', el, ve_defaults, setup_data)
        try:
            setup_data['cpu'] = convert_to_kernel_format(get_child_tag_atrr(el, tag='cpu', attr='limit'), lncpu=lncpu)
        except (ValueError, IndexError, TypeError):
            pass
        if setup_data['cpu'] is None:
            setup_data['cpu'] = ve_defaults['cpu']

check_value('io', el, ve_defaults, setup_data)

if ubc == 'true':
            check_value('mem', el, ve_defaults, setup_data)
        else:
            setup_data['mem'] = 0

try:
            setup_data['ep'] = int(get_child_tag_atrr(el, tag='other', attr='maxentryprocs'))
        except (ValueError, IndexError, TypeError):
            pass

check_value('nproc', el, ve_defaults, setup_data)
        check_value('pmem', el, ve_defaults, setup_data)
        check_value('iops', el, ve_defaults, setup_data)

else:
        # apply default limits
        prepare_setup_data(plan_id, reseller=reseller, lve_id=lve_id)
        if ubc == 'false':
            setup_data['mem'] = 0

if not result:
        cache_effective_limits(lve_id)
        if lve_id != 0 and _is_domain_isolated_user(lve_id):
            # Isolated user: apply the same limits everywhere so that
            # lvectl list and lvectl list-reseller stay in sync:
            #   lvp<uid>  — hard cap (total resources across user LVE + all domain LVEs)
            #   lve<uid>  — user's own LVE
            #   default   — per-LVE default inside lvp<uid> (what lvctl list-reseller shows)
            lve_setup(lve_id, lvp_id=lve_id)  # set lvp<uid> hard cap
            lve_setup(lve_id)                  # set lve<uid>

try:
                # Build domain→id mapping once, reuse for both mapping and
                # limit application to avoid redundant resolve_docroot calls.
                # _build_domain_map loads ~/.lve/domains.json under dropped
                # privileges; a broken pw entry or unreadable home would
                # otherwise abort the whole `lvectl apply all` (CLOS-4370).
                domain_map = lve._build_domain_map(lve_id)

# Map domain LVEs under the user LVP so they exist before
                # lve_set_default resets all per-LVE entries to user limits.
                lve._map_domain_lves(lve_id, lve.proc.map(), domain_map=domain_map)

pylve.lve_set_default(lve_id, lve_settings, err_msg=f'Can`t setup default settings for LVP {lve_id}')
                lve.apply_domain_lve_limits(lve_id, domain_map=domain_map)
            except (OSError, LvdError, PyLveError) as exc:
                syslog.syslog(
                    syslog.LOG_WARNING,
                    f"lve_apply: skipping per-domain isolation for uid={lve_id}: {exc}",
                )
        else:
            # apply limits
            lve_setup(lve_id, lvp_id=lvp_id)

def _pprint(*fields):
    """
    Print data with the last column 30 symbols wide.
    Useful for printing data that contains package names.
    """
    formatted_string = _format_fields(fields, wide_indices=[len(fields) - 1])
    print(formatted_string)

def _pprint_f(*fields):
    """
    Print data with the two last columns 30 symbols wide.
    Useful for printing full data of every user with package name
    and reseller name.
    """
    formatted_string = _format_fields(
        fields,
        wide_indices=[len(fields) - 2, len(fields) - 1],
    )
    print(formatted_string)

def _pprint_p(*fields):
    """
    Print data with the first column 30 symbols wide.
    Useful for printing packages data.
    """
    formatted_string = _format_fields(fields, wide_indices=[0])
    print(formatted_string)

def _pprint_r(*fields):
    """
    Print data with the first and last columns 30 symbols wide.
    Useful for printing data with user names and package names.
    """
    formatted_string = _format_fields(fields, wide_indices=[0, len(fields) - 1])
    print(formatted_string)

def _format_fields(fields: tuple, wide_indices: list, width: int = 30) -> str:
    """
    Helper function to format fields
    based on specified indices for wide columns.

Args:
        fields: The fields to format.
        wide_indices: List of indices in the fields that should be wide.
        width: The width of the wide columns.

Returns:
        A formatted string with specified fields widened.
    """
    formatted_fields = []

for index, field in enumerate(fields):
        # Convert field to string to ensure compatibility with formatting
        field_str = str(field)

if index in wide_indices:
            formatted_fields.append(f"{field_str:>{width}}")
        else:
            formatted_fields.append(f"{field_str:>8}")

return ''.join(formatted_fields)

def _pmem_vmem_to_bytes_value(value):
    """
    Convert pmem or vmem limits to bytes value

:param value: pmem or vmem limits in kbytes value
    :return: bytes value of limit

"""
    # if value was changed we remove asterisk from value for counted this
    value = str(value)
    was_changed = isinstance(value, str) and value.startswith('*')
    value = value.replace('*', '')
    if value:
        value = int(value)
        value *= 4096
    else:
        value = 0
    # return asterisk in value if this was changed
    value = f'*{value}' if was_changed else value
    return value

def _mb_mem(value):
    """
    Convert amount of RAM to M format

:param string value: amount of memory in KB
    :rtype: string
    :return: amount of memory in MB like "1234M"
    """
    result = ''
    was_changed = False
    if isinstance(value, str) and value.startswith('*'):
        value = value[1:]
        was_changed = True
    try:
        v = int(value)
    except ValueError:
        return ""
    if was_changed:
        result = '*'
    value = v * 4 // 1024
    if value > 0:
        result = f'{result}{value}M'
    else:
        result = f'{result}{v * 4}K'
    return result

def _formatter(printer, default_id="0", default_package="default", more_fields=None):
    """
    Generate header and default package data either as print to stdout or as json string
    """
    defaults = ve_defaults.copy()
    #  convert from kernel format for output
    defaults['cpu'] = defaults['cpu'] // 100

def get_data(key):
        return defaults.get(key, '')

_cpu = speed_to_old_cpu(get_data("cpu")) if get_data("cpu") != '' else ''

def convert_mem_limits(value):
        return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value)

fields_map = {
        'ID': default_id, 'SPEED': str(get_data('cpu')),
        'CPU': str(_cpu), 'NCPU': str(get_data('ncpu')),
        'PMEM': str(convert_mem_limits(get_data('pmem'))), 'VMEM': str(convert_mem_limits(get_data('mem'))),
        'EP': str(get_data('ep')), 'NPROC': str(get_data('nproc')), 'IO': str(get_data('io')),
        'IOPS': str(get_data('iops')), 'PACKAGE': default_package
    }
    res = []
    fields = get_fields()
    if more_fields is not None:
        fields += more_fields
    if JSON:
        line = ','.join(f'"{f}":"{fields_map.get(f, "")}"' for f in fields)
        res = [f'{{{line}}}']
    else:
        printer(*fields)
        printer(*[fields_map.get(f, "") for f in fields])
    return res

def _user_formatter(fields, printer=_pprint):
    """
    Generate inner function with closured fields names and printer function

:param list fields: List of strings that represent names of fields in final output
    :param callable printer: Function to format and print data for every entry
    :rtype: callable
    :return: function to format data for every user
    """
    def wrapper(user):
        """
        :param string user: Find and format data for this User ID
        :rtype: list
        :return: List of given user's statistics data line or empty list
        """
        data = ''
        package = packages_users[user]["package"]
        reseller = packages_users[user]["reseller"]
        if reseller == '':
            reseller = None
        if ve_cfg_version <= 1:
            # Reseller's default limits will not be inherited by its end-users.
            # Backward compatibility - show some reseller packages in
            # paneluserslimits.
            prepare_setup_data(package, reseller=None)
        else:
            if reseller is not None:
                # We can set xml_config_load_elements=False
                # because get_XML_cfg was called with True before _user_formatter called
                _load_resellers_xml_data(reseller, xml_config_load_elements=False)
            else:
                # It's important to re-read admin's limits here or we will use
                # limits from previous reseller for next users in list
                # We can set xml_config_load_elements=False
                # because get_XML_cfg was called with True before _user_formatter called
                get_XML_cfg(load_config_elements=False)
            prepare_setup_data(package, reseller=reseller)
        data = copy.copy(setup_data)        # only after reading reseller's xml
        lve_apply(user, plan_id=package, reseller=reseller, result=True)

def check_changed(key):
            return '*' + str(setup_data[key]) if str(data[key]) != str(setup_data[key]) else str(data[key])

def convert_mem_limits(value):
            return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value)

data['id'] = str(user)
        data['cpu'] = str(check_changed('cpu'))
        data['ncpu'] = str(check_changed('ncpu'))
        data['pmem'] = str(convert_mem_limits(check_changed('pmem')))
        data['vmem'] = str(convert_mem_limits(check_changed('mem')))
        data['ep'] = str(check_changed('ep'))
        data['io'] = str(check_changed('io'))
        data['nproc'] = str(check_changed('nproc'))
        data['iops'] = str(check_changed('iops'))
        if JSON:
            data['package'] = _normalize_str(package)
        else:
            data['package'] = package
        if reseller is None:
            data['reseller'] = 'N/A' if JSON else ''
        else:
            data['reseller'] = reseller

if '*' in data['cpu']:
            data['cpu'] =   '*' + str(int(data['cpu'].lstrip('*')) // 100)
        else:
            data['cpu'] = int(data['cpu']) // 100
        data['speed'] = data['cpu']
        data['cpu'] = str(speed_to_old_cpu(data['speed']))

res = []
        if JSON:
            line = ','.join(f'"{f}":"{data[f.lower()]}"' for f in fields)
            res = [f'{{{line}}}']
        else:
            printer(*[data[f.lower()] for f in fields])
        return res
    return wrapper

# Show current user's limits for control panel
# 'lvectl paneluserslimits' or 'lvectl paneluserlimits lve_id'
def paneluserslimits(userid=None, reseller=None):
    get_XML_cfg()
    try:
        # create cache for userid_calls
        GetControlPanelUsers('list-users')
        # use explicit compare, because userid may be zero!
        # if userid == 0, then show only default limits
        # LU-374
        if userid is not None and userid:
            GetControlPanelUsers('userid', userid)
        # LU-530
        elif reseller is not None:
            GetControlPanelUsers('list-reseller-users', reseller=reseller)
        else:
            GetControlPanelUsers('list-users')
    except Exception:
        pass
    more_fields = ["PACKAGE"]
    result = _formatter(_pprint, more_fields=more_fields)
    fields = get_fields() + more_fields
    formatter = _user_formatter(fields)
    for user in packages_users:
        result += formatter(user)
    if JSON:
        print('{"data":[' + ','.join(result) + ']}')

def paneluserslist():
    # type: () -> List[Tuple[int, str, str]]
    """Get list of tuples[lve_id, reseller, package] from control panel"""
    GetControlPanelUsers('list-users')
    result = []
    for str_uid, payload in packages_users.items():
        result.append((int(str_uid), payload['reseller'], payload['package']))
    return result

def panelpackagesdict():
    # type: () -> Dict[str, List[str]]
    """Get dict of pairs[provider, list[package_name]] from control panel"""
    from clveconfig import DEFAULT_PROVIDER  # NOQA
    packages = {}

GetControlPanelUsers('list-packages')
    # admin's packages are already in bytes...
    packages[DEFAULT_PROVIDER] = list(packages_users.keys())
    GetControlPanelUsers('list-resellers-packages')
    # ..but we must convert reseller's package to bytes, because cl-summary
    # expects bytes and print warnings about unicode comparison
    packages.update(packages_users)

return packages

# lvectl all-user-list
def all_users_limits():
    """
    Implements lvectl all-user-list command

:return: None, prints result to stdout
    """
    get_XML_cfg()
    GetControlPanelUsers('list-users')

result = _formatter(_pprint_f, more_fields=["PACKAGE", "RESELLER"])
    fields = get_fields() + ["PACKAGE", "RESELLER"]

formatter = _user_formatter(fields, printer=_pprint_f)
    for user in packages_users:
        result += formatter(user)

if JSON:
        print('{"data":[' + ','.join(result) + ']}')

def _filtering_da_admins(ve_dict):
    """
    Filtering DirectAdmin's admins for `lvectl apply all` command

:param ve_dict: dict with LVE
    :return: filtering dict

"""

if cldetectlib.getCPName() == 'DirectAdmin':
        # get list of uids DirectAdmin's admins
        uids_da_admins = [pwd.getpwnam(user).pw_uid for user in admins()]
        ve_dict = {key: value for key, value in ve_dict.items() if key not in uids_da_admins}

return ve_dict

def prepare_apply_data(lvp_id=0):
    try:
        # update packages_users global dict
        GetControlPanelUsers()
        packages_users_ = dict(packages_users)
        if lvp_id:  # filter for apply lve top containers
            cfg_lvp_id_list = lve.map.name_map.id_list()
            packages_users_ = {k: v for k, v in packages_users.items() if k in cfg_lvp_id_list}
    except Exception:
        packages_users_ = {}

if lvp_id is True:
        node_list = ve_lvp
        id_list = lvp_list()
    else:
        node_list = ve_lve
        id_list = lve.proc.lve_id_list(lvp_id=lvp_id)

# get xml node for each lve_id
    # ve_dict is a local dict with lve_id and node with limit for lve_id
    # keys - int lve_id
    # data - xml node or None
    ve_dict = {}
    for node in node_list:
        ve_dict[get_ve_lve_user_uid(ve_lve_element=node)] = {'node' : node, 'reseller' : None}
    for lve_id in id_list:
        if lve_id not in ve_dict:
            # add lve_id for LVE that are not in ve.cfg
            ve_dict[lve_id] = {'node': None, 'reseller': None}

if (len(packages_users_) != 0):
        #  filtering addon admins DA.
        #  package_users contain only users, not addon admins DA
        ve_dict = _filtering_da_admins(ve_dict)
        for uid in packages_users_:
            if uid not in ve_dict:
                node = None
            else:
                node = ve_dict[uid]['node']
            # add lve_id for users that have package assigned
            pkg = packages_users_[uid]
            resellers = guess_reseller_by_package(pkg)
            if len(resellers) > 0:
                ve_dict[uid] = {'node' : node, 'reseller' : resellers[0]}
            else:
                ve_dict[uid] = {'node' : node, 'reseller' : None}
    return ve_dict

def lve_destroy_and_recreate_all():
    lve_ve_dict, lve_lvp_map = _get_lve_ve_dict_and_lvp_map()

remaning_alive_lves = set(lve.proc.lve_id_list())
    # NOTE: First we handle all LVEs not belonging to any LVP.
    for lve_id in lve_ve_dict.keys():
        if lve_lvp_map.get(lve_id, 0) != 0:
            # This code path will never be triggered if reseller limits are disabled
            remaning_alive_lves.discard(lve_id)
            continue
        if lve_id in remaning_alive_lves:
            lve.lve_destroy(lve_id)
            remaning_alive_lves.discard(lve_id)
        # TODO(vlebedev): Is this context manager really necessary here?
        with lve.py.context_ignore_error(ignore_error=True):
            lve_apply(
                lve_id,
                out_node=lve_ve_dict[lve_id]['node'],
                reseller=lve_ve_dict[lve_id]['reseller'],
            )
    # NOTE: Reset the default LVE settings.
    lve_apply(lve_id=0)

if lve.reseller_limit_supported():
        lvp_ve_dict = prepare_apply_data(True)
        remaining_alive_lvps = set(lve.proc.lvp_id_list())

# NOTE: Now we handle LVPs and LVEs inside those LVPs.
        for lvp_id in lvp_ve_dict.keys():
            # Per-user isolation LVPs (lvp<uid>) are not resellers; skip them
            # like lve_apply_all() does.  This keeps reseller defaults from being
            # applied to them and avoids the CLOS-4513 -ESRCH: a nested isolation
            # LVP is destroyed as a side effect of destroying its parent reseller
            # LVP, so an explicit destroy here would hit an already-gone entry.
            # Discard so the cleanup loop below skips it too.
            if is_isolation_lvp(lvp_id):
                remaining_alive_lvps.discard(lvp_id)
                continue
            if lvp_id in remaining_alive_lvps:
                pylve.lve_lvp_destroy(lvp_id)
                remaining_alive_lvps.discard(lvp_id)
            _create_if_necessary_and_configure_lvp(lvp_ve_dict, lvp_id)

# apply reseller's users
            reseller_name = lve.map.get_reseller_name(lvp_id)
            kernel_mapping = lve.proc.map()
            for user in clcommon.cpapi.reseller_users(reseller_name):
                lve_id = pwd.getpwnam(user).pw_uid
                if lve_id in remaning_alive_lves:
                    lve.lve_destroy(lve_id)
                    remaning_alive_lves.discard(lve_id)

if kernel_mapping.get(lve_id, 0) != lvp_id:
                    lve.py.lve_lvp_move(
                        lvp_id,
                        lve_id,
                        err_msg=f'Can`t move lve_id={lve_id} to lvp_id={lvp_id}; error code {{code}}'
                    )
                lve_apply(lve_id=lve_id, reseller=reseller_name)

# NOTE: Destroy any remaining live LVP that was not handled above.
        for lvp_id in remaining_alive_lvps:
            # Skip per-user isolation LVPs: they are managed by the isolation
            # feature, and a nested one is already gone via its parent's cascade
            # (CLOS-4513) — destroying it here would raise -ESRCH.
            if is_isolation_lvp(lvp_id):
                continue
            pylve.lve_lvp_destroy(lvp_id)

# TODO(vlebedev): Not sure if this cleanup (and the one for LVPs above) is
    #                 really necesary but at least it faithfully replicates the
    #                 behaviour of `lvectl destroy all` part.
    # NOTE: Destroy any remaining live LVE that was not handled above.
    for lve_id in remaning_alive_lves:
        lve.lve_destroy(lve_id)

# Apply all users and resellers settings
def lve_apply_all():
    ve_dict, lve_lvp_map = _get_lve_ve_dict_and_lvp_map()
    domain_lve_ids = _get_all_domain_lve_ids()

with lve.py.context_ignore_error(ignore_error=True):
        for lve_id in ve_dict.keys():
            if lve_id in domain_lve_ids:
                # Domain LVEs are managed by apply_domain_lve_limits()
                # inside the _is_domain_isolated_user branch below.
                continue

if lve_lvp_map.get(lve_id, 0) == 0:
                lve_apply(lve_id, out_node=ve_dict[lve_id]['node'], reseller=ve_dict[lve_id]['reseller'])

lve_apply(lve_id=0)

# apply limits for all LVP and LVEs inside LVP
    if not lve.reseller_limit_supported():
        return

ve_dict = prepare_apply_data(True)
    kernel_mapping = lve.proc.map()

for lvp_id_ in ve_dict.keys():
        # User-level isolation LVPs (lvp<uid> for isolated users) are not
        # resellers; skip them here so reseller-default logic doesn't overwrite
        # the unlimited LVP defaults.  Use the marker-based is_isolation_lvp()
        # rather than _is_domain_isolated_user(): the kernel registers lvp<uid>
        # whenever isolation is allowed/enabled, even with 0 domains, and that
        # 0-domain case must be skipped too (CLOS-4513).
        if is_isolation_lvp(lvp_id_):
            continue

_create_if_necessary_and_configure_lvp(ve_dict, lvp_id_)

# apply reseller's users
        reseller_name = lve.map.get_reseller_name(lvp_id_)
        for user in clcommon.cpapi.reseller_users(reseller_name):
            lve_id_ = pwd.getpwnam(user).pw_uid

if _is_domain_isolated_user(lve_id_):
                # Create user LVP nested under the reseller LVP so the
                # hierarchy is correct from the start (no destroy-recreate
                # needed in _sync_map later).
                if not lve.proc.exist_lvp(lvp_id=lve_id_):
                    pylve.lve_lvp_create2(lve_id_, lvp_id_)

elif not lve.proc.check_inside_list(lve_id_, lvp_id_):
                    pylve.lve_lvp_destroy(lve_id_)
                    pylve.lve_lvp_create2(lve_id_, lvp_id_)

# User LVE goes under the user LVP, not the reseller LVP.
                if kernel_mapping.get(lve_id_, 0) != lve_id_:
                    lve.py.lve_lvp_move(
                        lve_id_,
                        lve_id_,
                        err_msg=f'Can`t move lve_id={lve_id_} to lvp_id={lve_id_}; error code {{code}}'
                    )
            else:
                # LU-511: create mapping before lve if needed
                if kernel_mapping.get(lve_id_, 0) != lvp_id_:
                    lve.py.lve_lvp_move(
                        lvp_id_,
                        lve_id_,
                        err_msg=f'Can`t move lve_id={lve_id_} to lvp_id={lvp_id_}; error code {{code}}'
                    )

lve_apply(lve_id=lve_id_, reseller=reseller_name)

def _get_lve_ve_dict_and_lvp_map() -> tuple[dict, dict]:
    get_XML_cfg()
    GetControlPanelUsers('list-users')
    lve_ve_dict = prepare_apply_data()
    lve_lvp_map = dict(lve.lve_id_lvp_id_pairs()) if lve.reseller_limit_supported() else {}
    return lve_ve_dict, lve_lvp_map

def _create_if_necessary_and_configure_lvp(ve_dict, lvp_id) -> None:
    # apply lvp limits using defaults for lvp
    # load reseller defaults instead of gloabal
    get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True, load_config_elements=False)
    lve_apply(lve_id=lvp_id,
              out_node=ve_dict[lvp_id]['node'],
              reseller=ve_dict[lvp_id]['reseller'],
              lvp_id=lvp_id)

get_XML_cfg(lvp_id=lvp_id, load_config_elements=False)
    lve_apply(lve_id=0, lvp_id=lvp_id)

def _remove_reseller(lvp_id):
    """Remove reseller from ve.cfg and from procfs.

CLOS-4419: idempotent for orphan kernel LVPs.  When no
    <reseller id="lvp_id"> block is present, the function falls through
    to a recovery branch that distinguishes three states:

* domain-isolated user — kernel LVP is legitimately maintained by
        _sync_map for the per-domain isolation feature; do not destroy.
      * already clean — neither block nor kernel LVP exist; return True
        so disable_reseller_limits reports OK instead of WARNING.
      * true orphan — block missing but kernel LVP exists with no
        isolation marker; destroy the LVP and re-parent any mapped
        LVEs to the host LVP.
    """
    get_global_lock(True)
    get_XML_cfg(lvp_id=lvp_id)
    for el in ve_lvp:
        if get_ve_lve_user_uid(ve_lve_element=el) == lvp_id:
            users = lve.proc.map_lve_id_list(lvp_id)

# Users with a per-user isolation LVP (lvp<uid>) have their LVE
            # nested inside that child LVP rather than directly under the
            # reseller LVP, so /proc/lve/map (and thus map_lve_id_list) does
            # not list them.  Add any such user to the users list so lve_apply()
            # rebuilds their LVP at root level and re-applies limits after the
            # reseller LVP is destroyed.
            #
            # CLOS-4542: use the marker-based is_isolation_lvp() — the original
            # semantics of this check (CLOS-3986 used _is_domain_isolated_user
            # when it too was marker-based).  CLOS-4370 later redefined
            # _is_domain_isolated_user to require >= 1 enabled domain, which
            # silently dropped bare-marker users (allow-domain-limits with no
            # enabled domains, e.g. after cagefsctl --isolates-allow-all) — they
            # also have a nested lvp<uid>, so being skipped here meant they were
            # cascade-destroyed with the reseller LVP and lost their limits.
            reseller_name = lve.map.get_reseller_name(lvp_id)
            for username in reseller_users(reseller_name):
                try:
                    uid = pwd.getpwnam(username).pw_uid
                except KeyError:
                    continue
                if is_isolation_lvp(uid) and uid not in users:
                    users.append(uid)

# move containers to host
            for lve_id in users[:]:
                pylve.lve_lvp_move(0, lve_id)
                try:
                    pwd.getpwuid(lve_id)
                except KeyError:
                    if lve.py.lve_exists(lve_id):
                        lve.lve_destroy(lve_id)
                    users.remove(lve_id)
            lvp_destroy(lvp_id)  # destroy container
            # NOTE: lvp_destroy raises PyLveError on kernel failure
            # (PyLve._wrapped_fun, ignore_error=False by default).  An
            # exception here propagates out before removeChild/save_xml
            # run, so ve.cfg cannot lose a <reseller> block while the
            # kernel still holds the LVP — that's the no-orphan invariant
            # pinned by test_remove_reseller_preserves_ve_cfg_on_destroy_failure.
            el.parentNode.removeChild(el)  # remove record
            save_xml(ve_cfg)
            # load defaults host settings for end users
            get_XML_cfg(lvp_id=0)
            # apply limits from config (including package limits)
            for lve_id in users:
                lve_apply(lve_id)
            get_XML_cfg(lvp_id=lvp_id)
            return True

# CLOS-4419 recovery branch.  No <reseller id="lvp_id"> block.
    if lvp_id <= 0:
        # Never touch the host LVP.
        return False
    if lvp_id in set(ws_config.find_all_lve_ids_with_config()):
        # Per-user domain-isolation LVP — _sync_map maintains this
        # kernel state on purpose.  Destroying it would just churn
        # (re-created on the next apply) and break isolation in the
        # meantime.  Surfacing it through list-reseller is handled
        # separately by the read-side filter in lvectl::parseLVPList.
        return False
    if not lve.proc.exist_lvp(lvp_id=lvp_id):
        # Neither block nor kernel LVP exist — already cleaned up
        # (e.g. by an earlier remove_absent_resellers tick).  The
        # caller (disable_reseller_limits) used to interpret False
        # here as anomaly and print
        # `{"status":"WARNING","msg":"no configuration found for LVP <id>"}`,
        # which is what the customer in Zendesk 283279 saw after the
        # 5-min cron beat them to the cleanup.  Treat as idempotent
        # success.
        return True

# True orphan: kernel LVP without a matching <reseller> block and
    # without a domain-isolation marker.  Re-parent any LVEs nested
    # inside it to the host LVP, destroy the LVP, and re-apply host
    # limits to the surviving users.
    users = lve.proc.map_lve_id_list(lvp_id)
    for lve_id in users[:]:
        pylve.lve_lvp_move(0, lve_id)
        try:
            pwd.getpwuid(lve_id)
        except KeyError:
            if lve.py.lve_exists(lve_id):
                lve.lve_destroy(lve_id)
            users.remove(lve_id)
    lvp_destroy(lvp_id)
    get_XML_cfg(lvp_id=0)
    for lve_id in users:
        lve_apply(lve_id)
    return True

def enable_domain_isolation(lve_id):
    """Enable per-domain LVE isolation for the given user.

Creates an empty id_registry file (marking the user as domain-isolated),
    creates lvp<uid> (nested under the user's reseller LVP if any), and
    moves lve<uid> under it.  Analogous to set-reseller for reseller limits.

Any existing domain IDs in the registry are pre-mapped to lvp<uid> via
    lve_lvp_map so that subsequent lve_setup calls will place domain LVEs
    under the correct LVP automatically.
    """
    ws_id_registry.create_empty_registry(lve_id)
    parent_lvp = lve.lve2lvp(lve_id)

if not lve.proc.exist_lvp(lvp_id=lve_id):
        pylve.lve_lvp_create2(lve_id, parent_lvp)
    elif parent_lvp and not lve.proc.check_inside_list(lve_id, parent_lvp):
        pylve.lve_lvp_destroy(lve_id)
        pylve.lve_lvp_create2(lve_id, parent_lvp)

kernel_mapping = lve.proc.map()
    if kernel_mapping.get(lve_id, 0) != lve_id:
        lve.py.lve_lvp_move(
            lve_id, lve_id,
            err_msg=f'Can`t move lve_id={lve_id} to lvp_id={lve_id}; error code {{code}}'
        )

# Pre-register LVP mapping for any domain IDs that already exist in the
    # registry (e.g. re-enabling after a previous disable).  lve_lvp_map
    # works for non-existent LVEs, so when lve_setup later creates them
    # the kernel places them under lvp<uid> automatically.
    for domain_id in ws_id_registry.get_all_entries(lve_id).values():
        if kernel_mapping.get(domain_id, 0) != lve_id:
            pylve.lve_lvp_map(lve_id, domain_id)

def disable_domain_isolation(lve_id):
    """Disable per-domain LVE isolation for the given user.

Moves lve<uid> back to the parent LVP (reseller or root), destroys
    domain LVEs, destroys lvp<uid>, and removes the id_registry file.
    Analogous to remove-reseller for reseller limits.
    """
    if not lve.proc.exist_lvp(lvp_id=lve_id):
        return

parent_lvp = lve.lve2lvp(lve_id)

# Move all LVEs out of the user LVP (user LVE + domain LVEs)
    for child_lve in lve.proc.map_lve_id_list(lve_id):
        pylve.lve_lvp_move(parent_lvp, child_lve)

kernel_mapping = lve.proc.map()
    if kernel_mapping.get(lve_id, 0) == lve_id:
        pylve.lve_lvp_move(parent_lvp, lve_id)

# Destroy domain LVEs
    for domain_id in ws_id_registry.get_all_entries(lve_id).values():
        if lve.proc.check_inside_list(domain_id):
            lve.lve_destroy(domain_id)

lvp_destroy(lve_id)
    ws_id_registry.remove_all_entries(lve_id)

def enable_domain_lve(owner_uid, docroot):
    """Register a single domain LVE under the user's LVP.

Assigns a domain ID for *docroot* in the per-user registry, then
    pre-maps it to *owner_uid*'s LVP so that lve_setup will place the
    domain LVE under the correct LVP when it is first created.

Also records the domain name in ``~/.lve/domains.json`` so we can obtain list of
    enabled domains for user.
    """
    domain_id = ws_id_registry.assign_domain_id(owner_uid, docroot)
    get_XML_cfg()
    proc_map = lve.proc.map()
    if proc_map.get(domain_id) != owner_uid:
        lve.py.lve_lvp_map(owner_uid, domain_id)
    ws_config.LvdConfig.load(owner_uid).add_domain_by_docroot(docroot)

def disable_domain_lve(owner_uid, docroot):
    """Unregister a single domain LVE and remove its registry entry.

Destroys the kernel LVE (if it exists) and removes *docroot* from
    the per-user domain ID registry.  Also removes the corresponding
    domain from ``~/.lve/domains.json``.
    """
    domain_id = ws_id_registry.get_domain_id(owner_uid, docroot)
    if domain_id is not None:
        get_XML_cfg()
        if lve.py.lve_exists(domain_id):
            try:
                lve.lve_destroy(domain_id)
            except PyLveError as exc:
                syslog.syslog(syslog.LOG_WARNING,
                              f"disable_domain_lve: failed to destroy "
                              f"lve {domain_id}: {exc}")
        ws_id_registry.remove_domain_id(owner_uid, docroot)
    ws_config.LvdConfig.load(owner_uid).remove_domain_by_docroot(docroot)

def disable_reseller_limits(reseller_name, lvp_id):
    """Disable reseller limits and call hooks"""
    if _remove_reseller(lvp_id):
        reseller_limits_disabled_post.throw_event(reseller=reseller_name)
    else:
        if JSON:
            json_format('multi', ['WARNING', f'no configuration found for LVP {lvp_id}'])
            sys.exit(-1)
        else:
            print(f'warning: no configuration found for LVP {lvp_id}')

# Delete User from ve.cfg and set default lve settings
def lve_delete(lve_id):
    _reject_if_domain_lve_id(lve_id)
    get_global_lock(True)
    get_XML_cfg()
    Deleted = False
    for el in ve_lve:
        if get_ve_lve_user_uid(ve_lve_element=el) == lve_id:
            Deleted = True
            lve_destroy(lve_id)
            lve_create(lve_id)
            el.parentNode.removeChild(el)
            save_xml(ve_cfg)
            get_XML_cfg()
            lve_apply(lve_id)
    if not Deleted:
        if JSON:
            json_format('multi', ['WARNING', f'no configuration found for VE {lve_id}'])
            sys.exit(-1)
        else:
            print(f'warning: no configuration found for VE {lve_id}')

def lve_enter_check():
    if not os.path.exists('/proc/lve/enter'):
        if JSON:
            json_format('multi', ['WARNING', 'enter by name not supported'])
        else:
            print('warning: enter by name not supported')
        sys.exit(-1)

def enter_apply(sign, binary):
    lve_enter_check()
    try:
        msg = sign + binary.strip()
        with open('/proc/lve/enter', 'w', encoding='utf-8') as f:
            f.write(msg)
    except Exception:
        pass

def list_binaries():
    get_XML_cfg()
    if JSON:
        result = '{"data":['
        first = True
        for el in ve_binary:
            path = el.getAttribute('path')
            if first:
                result += '"' + path + '"'
                first = False
            else:
                result += ',"' + path + '"'
        result += ']}'
        print(result)
    else:
        print("Binaries")
        for el in ve_binary:
            print(el.getAttribute('path'))

def load_binaries():
    get_XML_cfg()
    for el in ve_binary:
        enter_apply('+', el.getAttribute('path'))

def reload_binaries():
    lve_enter_check()
    with open('/proc/lve/enter', 'r', encoding='utf-8') as f:
        for line in f:
            enter_apply('-', line)
    load_binaries()

def del_binary(binary):
    global ve_binary
    get_global_lock(True)
    lve_enter_check()
    get_XML_cfg()
    deleted = False
    for el in ve_binary:
        if el.getAttribute('path') == binary:
            deleted = True
            enter_apply('-', binary)
            el.parentNode.removeChild(el)
            save_xml(ve_cfg)
            get_XML_cfg()
    if not deleted:
        if JSON:
            json_format('multi', ['WARNING', f'no configuration found for {binary}'])
        else:
            print(f'warning: no configuration found for {binary}')
        sys.exit(-1)

def set_binary(binary):
    global ve_binary
    global ve_enter_by_name
    get_global_lock(True)
    get_XML_cfg()
    for el in ve_binary:
        if el.getAttribute('path') == binary:
            return  # nothing to do, it is already there
    enter_apply('+', binary)
    bin_xml = ve_cfg.createElement('binary')
    bin_xml.setAttribute('path', binary)
    ve_enter_by_name.appendChild(bin_xml)
    save_xml(ve_cfg)
    get_XML_cfg()

def lve_set_default(set_data, package_flag, is_needed, lvp_id=0):
    """
    Set given lve or package to default values for given parameters

:param dict set_data: Arguments of lvectl call
    :param bool package_flag: Should we delete package or lve with given id
    :param callable is_needed: Function that takes xml element and set_data dict and returns
                               whether current xml element contains info about needed ID from set_data
    """
    try:
        if package_flag:
            data = ve_package
        elif lvp_id:
            data = ve_lvp
        else:
            data = ve_lve
        el = [e for e in data if is_needed(e, set_data)][0]
    except IndexError:
        return
    if lvp_id:
        # for lvectl set-reseller {id} --default=A,B,C; remove limit record in ve.cfg
        for tag_ in set_data['set-default']:
            if tag_ == 'ep':
                n = xml_filter_first(el, 'other', 'maxentryprocs')
            else:
                n = xml_filter_first(el, tag_, 'limit')
            if n:
                n.parentNode.removeChild(n)
        return
    to_keep = set(LIMITS_LIST_NAME) - set_data['set-default']
    for limit in to_keep:
        if limit == 'ep' and len(el.getElementsByTagName('other')) > 0:
            # dict.setdefault isn't lazy evaluated
            if limit not in set_data:
                set_data[limit] = el.getElementsByTagName('other')[0].getAttribute('maxentryprocs')
        elif len(el.getElementsByTagName(limit)) > 0:
            # dict.setdefault isn't lazy evaluated
            if limit not in set_data:
                set_data[limit] = el.getElementsByTagName(limit)[0].getAttribute('limit')
    # delete this lve or package
    if package_flag:
        plan_delete(set_data['ve_id'])
    else:
        lve_delete(set_data['ve_id'])

def _check_reseller_user_pair(uid, reseller_name):
    """
    Checks is uid owned by reseller
    :param uid: uid for check
    :param reseller_name: Reseller name, None treats as root
    :return: True - valid reseller/user pair, False - else
        Special case:
          if reseller_name is None (root) - always valid
    """
    if reseller_name in (None, 'root'):
        return True
    # reseller is not root
    # determine username
    username = get_main_username_by_uid(uid)
    if username in ('root', 'N/A'):
        # user is root or no such user -- error
        return False
    # determine users of supplied reseller's container.
    try:
        # Get reseller's users list
        reseller_users_list = reseller_users(reseller_name)
    except Exception:
        # any error - ignore, reseller is root
        reseller_users_list = []
    if username in reseller_users_list:
        return True
    return False

# Set limits for user
# TODO: split this method into several independent:
# - enable_reseller_limits
# - set_reseller_default_limits
# - set_reseller_limits
# - set_lve_limits
def lve_set(set_data, lvp_id=0):
    # set_data example:
    #  {'iops': 2222, 'reseller_name': 'res', 'save': False, 've_id': 1023, 'pmem': 524288}
    # 524288 * 4096 = 2G -- pmem=2G
    if lvp_id == 0 and set_data['ve_id'] != 0:
        _reject_if_domain_lve_id(set_data['ve_id'])
    if lvp_id == 0:
        # Set limits for user's LVE, check reseller/user match
        reseller_name = set_data.get('reseller_name', None)
        lve_id = set_data['ve_id']
        if not _check_reseller_user_pair(lve_id, reseller_name):
            return False
    global setup_data
    get_global_lock(True)
    if lvp_id and lvp_id == set_data['ve_id']:
        # reseller's container limits
        if lve.proc.exist_lvp(lvp_id):
            # reseller's container exists... load info about his container
            get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True)
        else:
            # reseller's container does not exists... create new one with default limits
            get_XML_cfg(lvp_id=lvp_id)
    elif lvp_id == 0 and lve.reseller_limit_supported():
        # user's limits (reseller & not)
        get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id']))
    else:
        # default limits and limits for user when reseller's does not supported
        get_XML_cfg(lvp_id=lvp_id)

try:
        GetControlPanelUsers()
    except Exception:
        pass
    try:
        # LU-366. Fix reset user's limits to unlimited if user in reseller package
        # and reseller is not root/admin
        package = packages_users[set_data['ve_id']]
        resellers = guess_reseller_by_package(package)
        reseller = resellers[0] if resellers else ''
        prepare_setup_data(package, reseller=reseller)
    except Exception:
        setup_data = ve_defaults
    if set_data['ve_id'] != 0:
        has_ve = False
        # set default

def is_needed_user(el, set_data):
            return get_ve_lve_user_uid(ve_lve_element=el) == set_data['ve_id']

if 'set-default' in set_data:
            lve_set_default(set_data, package_flag=False, is_needed=is_needed_user, lvp_id=lvp_id)
        if 'ncpu' in set_data:
            lncpu = int(set_data['ncpu'])
        else:
            lncpu = ve_defaults['ncpu']

# check that cpu value in any format (cpu, speed=% or speed=[m|g]hz) is equal or not
        cpu_is_different = True
        if 'cpu' in set_data:
            setted_cpu = convert_to_kernel_format(set_data['cpu'], lncpu = lncpu)
            if setted_cpu == setup_data['cpu']:
                cpu_is_different = False

if lvp_id:
            el_list = ve_lvp
        else:
            el_list = ve_lve  # choose top level container for modifications
        for el in el_list:
            if is_needed_user(el, set_data):
                for key in LIMITS_LIST_NAME:
                    if key in set_data:
                        try:
                            if key == 'ep':
                                set_child_tag_atrr(el, 'other', 'maxentryprocs', set_data[key])
                            else:
                                set_child_tag_atrr(el, key, 'limit', set_data[key])
                        except (ValueError, IndexError, TypeError):
                            # we already checked cpu value, so use cpu_is_different result
                            if key == "cpu":
                                is_different = cpu_is_different
                            # otherwise compare with default in usual way
                            else:
                                is_different = setup_data[key] != set_data[key]
                            if is_different or set_data['save']:
                                if key == 'ep':
                                    node = ve_cfg.createElement('other')
                                    node.setAttribute('maxentryprocs',str(set_data[key]))
                                else:
                                    node = ve_cfg.createElement(key)
                                    node.setAttribute('limit',str(set_data[key]))
                                el.appendChild(node)

if not set_data.get('skip-update-cfg', False):
                    save_xml(ve_cfg)

has_ve = True
                if lvp_id and lvp_id == set_data['ve_id']:
                    # reseller's container limits
                    if lve.proc.exist_lvp(lvp_id):
                        # reseller's container exists... load info about his container
                        get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True)
                    else:
                        # reseller's container does not exists... create new one with default limits
                        get_XML_cfg(lvp_id=lvp_id)
                elif lvp_id == 0 and lve.reseller_limit_supported():
                    # user's limits (reseller & not)
                    get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id']))
                else:
                    # default limits and limits for user when reseller's does not supported
                    get_XML_cfg(lvp_id=lvp_id)
                lve_apply(set_data['ve_id'], lvp_id=lvp_id)
            else:
                pass

if not has_ve and set_data['ve_id']:
            el_name = LVP_XML_TAG_NAME if lvp_id else 'lve'
            el = ve_cfg.createElement(el_name)
            if lvp_id:  # for create resellers limit config
                # set reseller_name reseller_id map
                el.setAttribute('id', str(lvp_id))
                el.setAttribute('user', set_data['user'])
                el.appendChild(ve_default)  # copy default limits to reseller
            else:
                if set_data.get('save-username'):
                    el.setAttribute('user', pwd.getpwuid(set_data['ve_id']).pw_name)
                else:
                    el.setAttribute('id', str(set_data['ve_id']))

for key in LIMITS_LIST_NAME:
                if key in set_data:
                    # we already checked cpu value, so use cpu_is_different result
                    if key == "cpu":
                        is_different = cpu_is_different
                    # otherwise compare with default in usual way
                    else:
                        is_different = setup_data[key] != set_data[key]
                    if is_different or set_data['save']:
                        if key == 'ep':
                            node = ve_cfg.createElement('other')
                            node.setAttribute('maxentryprocs',str(set_data[key]))
                        else:
                            node = ve_cfg.createElement(key)
                            node.setAttribute('limit',str(set_data[key]))
                        el.appendChild(node)

added = False
            for el2 in ve_package:
                el2.parentNode.insertBefore(el,el2)
                added = True
                break
            if not added:
                ve_cfg.lastChild.appendChild(el)

if not set_data.get('skip-update-cfg', False):
                save_xml(ve_cfg)

if lvp_id:
                enables_reseller_limits = not lve.proc.exist_lvp(lvp_id)
                # load lvp defaults and lvp tag, set limits for reseller
                get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True)
                lve_apply(set_data['ve_id'], lvp_id=lvp_id)
                # copy default limits from host container
                pylve.lve_set_default(lvp_id, pylve.lve_info(0))
                # load lvp tag and reseller's end user defaults
                get_XML_cfg(lvp_id=lvp_id)
                reseller_name = lve.map.get_reseller_name(lvp_id)
                for lve_id_ in lve.map.lvp_lve_id_list(lvp_id=lvp_id):
                    lve.py.lve_lvp_move(lvp_id, lve_id_)
                    lve_apply(lve_id_, reseller=reseller_name)
                # call hook if we enabled reseller limits
                if enables_reseller_limits:
                    reseller_limits_enabled_post.throw_event(reseller=reseller_name)
            else:
                if lve.reseller_limit_supported():
                    get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id']))
                else:
                    get_XML_cfg(lvp_id=lvp_id)
                lve_apply(set_data['ve_id'])
    else:
        for key in LIMITS_LIST_NAME:
            if key in set_data:
                if key == 'ep':
                    ve_default.getElementsByTagName('other')[0].setAttribute('maxentryprocs',str(set_data[key]))
                else:
                    ve_default.getElementsByTagName(key)[0].setAttribute('limit',str(set_data[key]))

if not set_data.get('skip-update-cfg', False):
            save_xml(ve_cfg)

get_XML_cfg(lvp_id=lvp_id)
        lve_apply(set_data['ve_id'], lvp_id=lvp_id)
    return True

def package_set(set_data, is_reseller=False):
    """
    Set package with some heuristic algorithm to simulate old package set behavior
    """
    get_global_lock(True)
    get_XML_cfg()
    # Removed in LU-351
    # set_data['ve_id'] = unicode(set_data['ve_id'].decode('utf-8'))
    reseller_list = guess_reseller_by_package(set_data['ve_id'])
    if len(reseller_list) == 0:
        reseller = None
    elif len(reseller_list) >= 1:
        # if dublicated packages found - use first as a reseller
        reseller = reseller_list[0]
    #if ve.cfg has tag <version>2</version> - trying to guess reseller name by package
    #if reseller is undef or ve.cfg has not tag version - work with ver 1
    if reseller is not None and ve_cfg_version > 1:
        set_data['reseller_name'] = reseller
        package_set_ext(set_data, is_reseller=True)
    else:
        package_set_ext(set_data, is_reseller=False)

# Set new package or modify exist
# package-set-ext
def package_set_ext(set_data, is_reseller=False):
    get_global_lock(True)
    get_XML_cfg()
    has_package = False

if is_reseller:
        def is_needed_plan(el, set_data):
            return (
                el.getAttribute('id') == set_data['ve_id'] and el.getAttribute('reseller') == set_data['reseller_name']
            )
    else:
        def is_needed_plan(el, set_data):
            return el.getAttribute('id') == set_data['ve_id'] and not el.getAttribute('reseller')

if 'set-default' in set_data:
        lve_set_default(set_data, package_flag=True, is_needed=is_needed_plan)

for el in ve_package:
        if is_needed_plan(el, set_data):
            for key in LIMITS_LIST_NAME:
                if key in set_data:
                    try:
                        if key == 'ep':
                            el.getElementsByTagName('other')[0].setAttribute('maxentryprocs',str(set_data[key]))
                        else:
                            el.getElementsByTagName(key)[0].setAttribute('limit',str(set_data[key]))
                    except (ValueError, IndexError, TypeError):
                        if key == 'ep':
                            node = ve_cfg.createElement('other')
                            node.setAttribute('maxentryprocs',str(set_data[key]))
                        else:
                            node = ve_cfg.createElement(key)
                            node.setAttribute('limit',str(set_data[key]))
                        el.appendChild(node)
            if cldetectlib.is_plesk():
                plesk_id = _plesk_get_package_id(set_data.get('reseller_name', ''), set_data['ve_id'])
                el.setAttribute(XML_PLESK_ID, str(plesk_id))
            has_package = True
    if not has_package:
        package_reseller = ''
        el = ve_cfg.createElement('package')
        el.setAttribute('id', set_data['ve_id'])
        if is_reseller:
            el.setAttribute('reseller', set_data['reseller_name'])
            package_reseller = set_data['reseller_name']
        if cldetectlib.is_plesk():
            plesk_id = _plesk_get_package_id(package_reseller, set_data['ve_id'])
            el.setAttribute(XML_PLESK_ID, str(plesk_id))
        for key in LIMITS_LIST_NAME:
            if key in set_data:
                if key == 'ep':
                    node = ve_cfg.createElement('other')
                    node.setAttribute('maxentryprocs',str(set_data[key]))
                else:
                    node = ve_cfg.createElement(key)
                    node.setAttribute('limit',str(set_data[key]))
                el.appendChild(node)
        ve_cfg.lastChild.appendChild(el)

save_xml(ve_cfg)
    get_XML_cfg()
    copy_package_settings_to_cpanel(set_data)

if 'ncpu' in set_data:
        lncpu = int(set_data['ncpu'])
    else:
        lncpu = ve_defaults['ncpu']

if 'cpu' in set_data:
        set_data['cpu'] = convert_to_kernel_format(set_data['cpu'], lncpu=lncpu)

reseller = set_data['reseller_name'] if is_reseller else None
    plan_apply(set_data['ve_id'], reseller=reseller)

def _plesk_get_package_id(reseller: str, package: str) -> Optional[int]:
    """
    Find the right package id from plesk DB query
    """
    panel = detect_panelclass()
    packages = panel.list_domain_packages_with_id()
    try:
        pack = next(filter(
            lambda x: x[0] in {reseller, 'root'} and x[1] == package,   # no reseller == reseller is root (admin)
            packages
        ))
        return pack[2]
    except StopIteration:
        return None

def get_reseller_packages_map():
    """
    Retrives resellers to packages map from panel using /usr/bin/getcontrolpaneluserspackages
    :return: Dictionary:
    { 'reseller1' -> ['pack1', 'pack2'], 'reseller2' -> ['pack'] }
    """
    global packages_users
    packages_users_copy = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages_map = packages_users
    packages_users = packages_users_copy
    return reseller_packages_map

def reseller_package_set(set_data):
    """
    Set reseller package limits
    :param set_data: input data dictionary
    :return: True - limits was set succesfully
            False - supplied provider has no supplied package
    """
    # set limits to package that belongs to given reseller
    reseller_name = set_data['reseller_name']
    package_name = set_data['ve_id']
    # Retrive resellers packages from panel
    reseller_packages_map = get_reseller_packages_map()
    # If reseller has supplied package -- set limit
    if reseller_name in reseller_packages_map and package_name in reseller_packages_map[reseller_name]:
        # Reseller/package pair valid - set limit
        package_set_ext(set_data, is_reseller=True)
        return True
    # ERROR: Supplied reseller has no supplied package
    return False

def copy_package_settings_to_cpanel(set_data):
    """
    Copy package limits from ve.cfg to cpanel packages data
    """
    package = set_data['ve_id']
    if not cldetectlib.is_cpanel():
        return  # skip func if panel not cPanel
    package_path = f'/var/cpanel/packages/{package}'
    if not os.path.isfile(package_path):
        return  # skip func if no cPanel packages found
    with open(package_path, 'r', encoding='utf-8') as f:
        cpanel_package_data = f.readlines()
    new_cpanel_package_data = cpanel_package_data[:]
    old_cpanel_data = {}

# proces old_cpanel_package_data - get old limits and remove stings from it
    # result of processing - cpanel_package_data_modify and old_cpanel_data
    for line in cpanel_package_data:
        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]  # get old_limits
            if limit_name in LIMITS_LIST_NAME:
                new_cpanel_package_data.remove(line)
        if line.startswith('_PACKAGE_EXTENSIONS') and 'lve' not in line:
            return      # skip func if no lve extention install to the package

for limit_name in ('pmem', 'mem', 'vmem'):
        if limit_name in old_cpanel_data:
            memory_page_value = clcommon.memory_to_page(old_cpanel_data[limit_name])
            old_cpanel_data[limit_name] = memory_page_value or ve_defaults[limit_name]

if is_limits_equals(old_cpanel_data, set_data):
        return              # skip writeting to file - limits are equals

# create and add to new cpanel_data_file limits lines like:
    # lve_ + limit_name + = + limit_value
    cpanel_data = create_cpanel_limits(package, ve_package)
    for limit_name in LIMITS_LIST_NAME:
        limit_value = cpanel_data[limit_name]
        limit_line = f'lve_{limit_name}={limit_value}\n'
        new_cpanel_package_data.append(limit_line)
    write_file_via_tempfile(''.join(new_cpanel_package_data), package_path, 0o644)

def is_limits_equals(old_limits, new_limits):
    """
    check if new set of limits for package are equals to used
    """
    for key in new_limits.keys():
        if key in ('ve_id', 'save'):
            continue    # ve_id == package name. skip this key
        try:
            if old_limits[key] != new_limits[key]:
                return False
        except KeyError:
            return False
    return True

def create_cpanel_limits(package_id, xml_packages):
    """
    create limits for cpanel package file
    use data from ve.cfg:
    limit = limit if found in ve.cfg or DEFAULT
    return dict
    """
    result_data = {}
    for el in xml_packages:
        if el.getAttribute('id') == package_id:
            for limit in LIMITS_LIST_NAME:
                try:
                    if limit == 'ep':
                        result_data[limit] = str(
                            el.getElementsByTagName('other')[0].getAttribute('maxentryprocs')
                        ).strip()
                    elif limit in ("mem", "vmem", "pmem"):
                        result_data[limit] = str(
                            clcommon.page_to_memory(
                                int(el.getElementsByTagName(limit)[0].getAttribute("limit"))
                            )
                        ).strip()
                    else:
                        result_data[limit] = str(el.getElementsByTagName(limit)[0].getAttribute('limit')).strip()
                except (ValueError, IndexError, TypeError):
                    result_data[limit] = 'DEFAULT'
    return result_data

# Delete plan from ve.cfg
def plan_delete(plan_id, reseller_name=None):
    get_global_lock(True)
    get_XML_cfg()
    Deleted = False
    if reseller_name is None:
        def is_needed_package(el):
            return el.getAttribute('id') == plan_id and not el.getAttribute('reseller')
    else:
        def is_needed_package(el):
            return el.getAttribute('id') == plan_id and el.getAttribute('reseller') == reseller_name

for el in ve_package:
        if is_needed_package(el):
            Deleted = True
            el.parentNode.removeChild(el)
            save_xml(ve_cfg)
            break
    if not Deleted:
        # try to guess reseller name only if no reseller name
        if reseller_name is None:
            resellers_list = guess_reseller_by_package(plan_id)
            if len(resellers_list) == 0:
                reseller = None
            else:
                # if some resellers found - use first
                reseller = resellers_list[0]
            # try to delete package only if we guess reseller
            if reseller is not None:
                plan_delete(plan_id, reseller)
                return
        if JSON:
            json_format(
                'multi',
                ['WARNING', f'no configuration found for plan {plan_id}']
            )
            sys.exit(-1)
        else:
            print(f'warning: no configuration found for plan {plan_id}')
    lve_apply_all()

def reseller_plan_delete(plan_id, reseller_name):
    plan_delete(plan_id, reseller_name=reseller_name)

def get_xml_limit(el, key):
    try:
        return str(el.getElementsByTagName(key)[0].getAttribute('limit'))
    except (ValueError, IndexError):
        #  convert from kernel format to output
        return ve_defaults[key] if key != 'cpu' else f'{ve_defaults[key] // 100}%'

def _normalize_str(data_str):
    """
    Normalize string for JSON output.
    Example:
     - Input string: -_&[{}]'"`te\\s/t\a
     - Output string: -_&[{}]'\"`te\\\\s/t\\a
    :param data_str: String for normalize
    :return: Normalied string
    """
    def _get_char_index(input_string, char_to_search, ordinal):
        """
        Get the index of the specified occurrence of character in string
        :param input_string: String
        :param char_to_search: Character to search
        :param ordinal: Required occurence number
        :return: Char index
        """
        count = 0
        for idx, ch in enumerate(input_string):
            if ch == char_to_search:
                # Char found
                count += 1
                if count == ordinal:
                    return idx
        # Char not found
        return -1

if data_str is None:
        return None
    json_str = json.dumps({'str': data_str})
    # json_str example: {"str": "-_&[{}]'\"`te\\\\s/t\\a"},
    # get '-_&[{}]'\"`te\\\\s/t\\a' from it
    # Get third " index
    trd_idx = _get_char_index(json_str, '"', 3)
    # Get Last " index
    last_idx = json_str.rfind('"')
    return json_str[trd_idx + 1:last_idx]

def _package_formatter(fields, is_reseller=False, printer=None):
    """
    Generate inner function with closured fields names, is_reseller flag and printer function

:param list fields: List of strings that represent names of fields in final output
    :param boolean is_reseller: Format output with info about reseller or not
    :param callable printer: Function to format and print data for every entry
    :rtype: callable
    :return: function to format data for every user
    """

printer = printer if printer is not None else _pprint_r if is_reseller else _pprint_p

def wrapper(package_name, reseller_name=None):
        """
        :param string package_name: Find and format data for this package name
        :param string reseller_name: reseller name, owner of supplied package
        :rtype: list
        :return: List of giver package's statistics data line or empty list
        """
        # in order to avoid unicode warnings here
        if is_reseller:
            # Reseller package
            _load_resellers_xml_data(reseller_name)
            data = copy.copy(ve_defaults)   # only after reading reseller's xml
            data['reseller'] = reseller_name

def is_needed_package(el):
                return package_name == el.getAttribute('id') and el.getAttribute('reseller') == reseller_name
        else:
            # Admin's package
            def is_needed_package(el):
                return package_name == el.getAttribute('id') and not el.getAttribute('reseller')

data = copy.copy(ve_defaults)

data['id'] = _normalize_str(package_name) if JSON else package_name

def convert_mem_limits(value):
            return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value)

data['vmem'] = convert_mem_limits(data['mem'])
        if LVE_VERSION > 4:
            data['pmem'] = convert_mem_limits(data['pmem'])
        for el in ve_package:
            if is_needed_package(el):
                lncpu = get_xml_limit(el, 'ncpu')
                data['ncpu'] = lncpu if lncpu != '' else str(ve_defaults['ncpu'])
                data['speed'] = str(convert_to_kernel_format(
                    get_xml_limit(el, 'cpu'), lncpu=int(data['ncpu'])) // 100
                )
                try:
                    data['ep'] = str(int(
                        el.getElementsByTagName('other')[0].getAttribute('maxentryprocs')
                    ))
                except (IndexError, ValueError):
                    pass
                data['pmem'] = convert_mem_limits(get_xml_limit(el, 'pmem'))
                data['vmem'] = convert_mem_limits(get_xml_limit(el, 'mem'))
                data['io'] = get_xml_limit(el, 'io')
                data['nproc'] = get_xml_limit(el, 'nproc')
                data['iops'] = get_xml_limit(el, 'iops')
        if data.get('speed') is None:
            #  convert from kernel format for output
            data['speed'] = data['cpu'] // 100
        data['cpu'] = str(speed_to_old_cpu(data['speed']))

res = []
        if JSON:
            line = ','.join(f'"{f}":"{data.get(f.lower(), "N/A")}"' for f in fields)
            res = [f'{{{line}}}']
        else:
            printer(*[data.get(f.lower(), '') for f in fields])
        return res
    return wrapper

# lvectl package-list
def get_packages_list():
    get_XML_cfg()
    GetControlPanelUsers('list-packages')
    packages = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages = packages_users.copy()

result = _formatter(_pprint_p, default_id=DEFAULT_PACKAGE)
    formatter = _package_formatter(get_fields(), is_reseller=False, printer=_pprint_p)
    for package in packages:
        result += formatter(package)
    if ve_cfg_version > 1:
        formatter = _package_formatter(get_fields(), is_reseller=True, printer=_pprint_p)
    # reseller_packages: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in reseller_packages.items():
        # On DA skip all non-admin's packages
        if cldetectlib.is_da() and reseller_name != 'admin':
            continue
        for reseller_package in packages_list:
            if ve_cfg_version > 1:
                result += formatter(reseller_package, reseller_name)
            else:
                result += formatter(reseller_package)

if JSON:
        print('{"data":[' + ','.join(result) + ']}')

# lvectl reseller-package-list
def get_resellers_packages_list():
    get_XML_cfg()
    GetControlPanelUsers('list-resellers-packages')

more_fields = ["RESELLER"]
    fields = get_fields() + more_fields
    result = _formatter(_pprint_p, default_id=DEFAULT_PACKAGE)

formatter = _package_formatter(fields, is_reseller=True)
    # packages_users: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in packages_users.items():
        for reseller_package in packages_list:
            result += formatter(reseller_package, reseller_name)

if JSON:
        print('{"data":[' + ','.join(result) + ']}')

# lvectl all-package-list
def get_all_packages_list():
    get_XML_cfg()
    GetControlPanelUsers('list-packages')
    packages = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages = packages_users.copy()

more_fields = ["RESELLER"]
    fields = get_fields() + more_fields
    # make header with default package
    result = _formatter(_pprint_r, default_id=DEFAULT_PACKAGE, more_fields=more_fields)

formatter = _package_formatter(fields, is_reseller=False, printer=_pprint_r)
    for package in packages:
        result += formatter(package)
    # Print resellers packages
    formatter = _package_formatter(fields, is_reseller=True)
    # reseller_packages: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in reseller_packages.items():
        for reseller_package in packages_list:
            result += formatter(reseller_package, reseller_name)

if JSON:
        print('{"data": [' + ','.join(result) + ']}')

cached_resellers_packages = None
cached_list_packages      = None
cached_users              = None
cached_reseller_users     = None
cached_default            = None

def _convert_packages_list(package_list):
    """
    Converts package list to internal format
    :param package_list: Package list. Example: ['BusinessPackage', 'Package2']
    :return: Package list as dictionary. Example: {'BusinessPackage': 'BusinessPackage', 'Package2': 'Package2'}
    """
    packages_users_dict = {}
    for package in package_list:
        packages_users_dict[package] = package
    return packages_users_dict

# Get users from control panel with plans
def GetControlPanelUsers(option='list-all', lve_package_id='', reseller=None):
    """
    Parse output from GET_CP_PACKAGE_SCRIPT and get package and lve relations

:param option: option for GET_CP_PACKAGE_SCRIPT.
     Option is one from the following possible values:  'userid', 'package', 'list-packages', 'list-resellers-packages'
    :type option: string
    :param lve_package_id: lve_id or package_name
    :type lve_package_id: string or int
    :param reseller:
    :type reseller: string
    """
    global cached_list_packages
    global cached_resellers_packages
    global cached_users
    global cached_reseller_users
    global cached_default
    global packages_users
    # Check arguments
    if option not in ('list-all', 'userid', 'package', 'list-packages',
                      'list-resellers-packages', 'list-users', 'list-reseller-users'):
        return False
    if option in ('userid', 'package') and lve_package_id == '':
        return False

from clcommon.cpapi import (  # pylint: disable=import-outside-toplevel
        admin_packages,
        get_reseller_users,
        get_uids_list_by_package,
        list_all,
        list_users,
        reseller_package_by_uid,
        resellers_packages,
    )

try:
        if option == 'userid':
            if cached_users is not None:
                try:
                    packages_users = {
                        lve_package_id: {
                            'package': cached_users[lve_package_id]['package'],
                            'reseller': cached_users[lve_package_id]['reseller']
                        }
                    }
                except KeyError:
                    packages_users = {lve_package_id: {'package': '', 'reseller': ''}}
            else:
                try:
                    reseller_name, package = reseller_package_by_uid(lve_package_id)
                except ValueError:
                    # this is possible on vm without control panel
                    reseller_name = package = ''
                packages_users = {lve_package_id: {'package': package, 'reseller': reseller_name}}
                return True
        elif option == 'package':
            # reseller - optional argument
            packages_users = {lve_package_id: get_uids_list_by_package(lve_package_id, reseller)}
            return True
        elif option == 'list-packages':
            # Result format:
            # {'BusinessPackage': 'BusinessPackage', 'Package2': 'Package2'}
            if cached_list_packages is None:
                package_list = admin_packages()
                # Convert to output format
                packages_users = _convert_packages_list(package_list)
                cached_list_packages = packages_users
            else:
                # list-packages data already present
                packages_users = cached_list_packages
            return True
        elif option == 'list-resellers-packages':
            # Result format:
            #  {'res2 SimplePackage': 'res2 Package',
            #   'res1 BusinessPackage': 'res1 UltraPackage'}
            if cached_resellers_packages is None:
                # in order to produce same results as code that works with Popen
                packages_users = resellers_packages()
                cached_resellers_packages = packages_users
            else:
                # list-resellers-packages data already present
                packages_users = cached_resellers_packages
            return True
        elif option == 'list-users':
            # Result format:
            # {1000: {'reseller': '', 'package': 'Package1'},
            #  1001: {'reseller': '', 'package': 'BusinessPackage'},
            # }
            if cached_users is None:
                cached_users = list_users()
            packages_users = cached_users
            return True
        elif option == 'list-reseller-users':
            # {1001: {'reseller': 'res1', 'package': 'BusinessPackage'},
            #  1004: {'reseller': 'res1', 'package': 'BusinessPackage'}}
            if cached_reseller_users is None:
                reseller_users_dict = get_reseller_users(reseller)
                # for uid, user_data in reseller_users_dict.iteritems():
                #     packages_users[uid] = user_data
                cached_reseller_users = reseller_users_dict
                packages_users = reseller_users_dict
            else:
                # list-reseller-users data already present
                packages_users = cached_reseller_users
            return True
        elif option == 'list-all':  # deprecated. TODO: Remove this option
            # Result format:
            # {1000: 'Package1', 1001: 'BusinessPackage'}
            if cached_default is None:
                cached_default = list_all()
            packages_users = cached_default
            return True
    except EncodingError as e:
        raise_cpanel_encoding_error(e)
    except OSError:
        pass
    return False

def get_panel_users_count():
    """
    Retrieves panel users count
    :return:
    """
    GetControlPanelUsers()
    return len(packages_users)

# Apply plan settings for users
def plan_apply(plan_id, reseller=None):
    # fill the cache to speedup `lvectl package-set-ext` with many users in one package
    GetControlPanelUsers("list-users")

if GetControlPanelUsers("package", plan_id, reseller=reseller):
        for uid in packages_users[plan_id]:
            lve_apply(int(uid), plan_id, reseller=reseller)

# Destroy many LVEs from stdin
def destroy_many(users_list):
    for line in users_list:
        line = line.replace('\n','')
        users = line.strip().split()
        for user in users:
            if (len(user) != 0):
                try:
                    user = int(user)
                    lve_destroy(user)
                except Exception:
                    pass

# Apply many LVEs from stdin
def apply_many(users_list):
    get_XML_cfg()
    try:
        GetControlPanelUsers()
    except Exception:
        pass

for line in users_list:
        line = line.replace('\n','')
        users = line.strip().split()
        for user in users:
            if (len(user) != 0):
                try:
                    user = int(user)
                    lve_apply(user)
                except Exception:
                    pass

# Put pid into LVE
def limit_pid(lve_id, pid, flags):
    pylve.lve_enter_pid_flags(
        int(lve_id), int(pid), flags,
        err_msg=f'Can`t put proccess with pid {pid} in lve {lve_id}; error code {{code}}'
    )

# Get pid from LVE
def release_pid(pid):
    pylve.lve_leave_pid(int(pid), err_msg=f'Can`t release process with pid {pid}')

def get_globals():
    global ve_cfg
    global ve_lveconfig
    global ve_default
    global ve_lve
    global ve_defaults
    global ve_package
    global ve_binary
    global ve_enter_by_name
    global ubc

return {'ve_cfg': ve_cfg, 've_lveconfig': ve_lveconfig,
            've_default': ve_default,
            've_lve': ve_lve, 've_defaults': ve_defaults,
            've_package': ve_package, 'ubc': ubc,
            've_enter_by_name': ve_enter_by_name}

def guess_reseller_by_package(package):
    reseller = []
    global packages_users
    pkg_users_old = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages = packages_users.copy()
    packages_users = pkg_users_old.copy()

# reseller_packages: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in reseller_packages.items():
        for package_name_in_key in packages_list:
            if package == package_name_in_key:
                reseller.extend([reseller_name])
    return reseller

# LU-400
def call_endurance_custom_script(args):
    """
    Call Endurance's custom script

:param args: list of arguments for pass to Endurance's custom script
    :return: None

"""

endurance_custom_script = cldetectlib.get_param_from_file(cldetectlib.CL_CONFIG_FILE,
                                                              'ENDURANCE_CUSTOM_SCRIPT',
                                                              separator='=')
    if endurance_custom_script and os.path.isfile(endurance_custom_script):

ret_code, std_out = exec_utility(endurance_custom_script, args)
        if ret_code != 0:
            message = f'Error while executing Endurance\'s custom script\n{std_out}'
            if JSON:
                json_format('multi', ['ERROR', message])
            else:
                err_message = f"error: {message}"
                sys.stderr.write(f"{err_message}\n")
                sys.exit(ret_code)

def _page_to_memory_or_bytes(value):
    """
    Convert page value to human-readable value or bytes, depending on BYTES_FLAG;
    E.g.
    >>> _page_to_memory_or_bytes(1233254)  # BYTES_FLAG=False
    '100M'
    >>> _page_to_memory_or_bytes(1233254)  # BYTES_FLAG=True
    654321

:type value: int
    :rtype: str | int
    """
    if BYTES_FLAG:
        return int(round(value * mmap.PAGESIZE))
    return _mb_mem(value)

def remove_absent_resellers():
    """
    Remove from LVE all resellers, which are absent from panel
    :return: None
    """
    # Build resellers ids list for removal
    reseller_id_list_for_delete = []
    # Get resellers list from cpapi
    # Create a list from a generator for repeated membership testing
    cpapi_resellers_list = list(lve.map.resellers())
    get_XML_cfg()  # for loading reseller_name<=>reseller_id map form ve.cfg
    # If reseller present in LVE, but absent in panel - add it to list for removal
    for lve_reseller_id in lvp_list():
        try:
            lve_reseller_name = lve.map.get_reseller_name(lve_reseller_id)
            if lve_reseller_name not in cpapi_resellers_list:
                # Reseller does not exist in panel, remove it from LVE
                reseller_id_list_for_delete.append(lve_reseller_id)
        except (KeyError, OSError, IOError):
            # No such user, remove it from LVE
            reseller_id_list_for_delete.append(lve_reseller_id)
    # Remove all selected resellers ignoring all errors (for example no such reseller)
    for reseller_id_for_delete in reseller_id_list_for_delete:
        _remove_reseller(reseller_id_for_delete)

def regenerate_domains(uid, domain=None, old_domain=None, old_docroot=None):
    """Update LVD domain config, docroot-to-ID mapping, and kernel LVEs.

Called after a control-panel hook detects a domain rename,
    document-root change, or user rename.

Two modes of operation:

**Single-domain** (*domain* is given):
        1. Rename domain in ~/.lve/domains.json (if *old_domain* given).
           Uses privilege drop so files stay user-owned.
        2. Ensure the current docroot has a domain_id in the registry.
           If *old_docroot* is given, removes the old mapping and assigns
           a new domain_id for the new docroot.
        3. Destroy the old domain LVE in the kernel (if it existed).
        4. Map the new domain_id to the user's LVP and re-apply limits
           from domains.json so that per-domain limits keep working.

**All domains** (*domain* is ``None``):
        Iterates every domain in the user's ``domains.json``, detects
        stale registry entries whose docroot no longer matches (e.g.
        after a user rename shifted the home directory), and performs
        the single-domain reassignment for each one.

Args:
        uid: numeric user ID.
        domain: current (possibly renamed) domain name.  When ``None``,
            all domains from the user's config are checked.
        old_domain: previous domain name if it was renamed.
        old_docroot: previous document root path if it changed.
    """
    if domain is None:
        _regenerate_all_domains(uid)
        return

if old_domain:
        with ws_config.user_context(uid):
            config = ws_config.LvdConfig.load(uid)
            entry = config.find_domain(name=old_domain)
            if entry is not None:
                entry.name = domain
                config.save()

try:
        docroot = ws_config.resolve_docroot(domain)
    except Exception:
        return

registry = ws_id_registry.get_all_entries(uid)
    old_domain_id = None

if docroot not in registry:
        if old_docroot and old_docroot in registry:
            old_domain_id = registry[old_docroot]
            ws_id_registry.reassign_docroot(uid, old_docroot, docroot)

if old_domain_id is not None and lve is not None:
        if lve.proc.check_inside_list(old_domain_id, lvp_id=uid):
            pylve.lve_destroy(old_domain_id)

new_domain_id = ws_id_registry.get_domain_id(uid, docroot)
        if new_domain_id is not None:
            kernel_mapping = lve.proc.map()
            if kernel_mapping.get(new_domain_id, 0) != uid:
                pylve.lve_lvp_map(uid, new_domain_id)
            domain_map = lve._build_domain_map(uid)
            lve.apply_domain_lve_limits(uid, domain_map=domain_map)

def _regenerate_all_domains(uid):
    """Synchronise the domain-ID registry with current docroots.

Compares docroots currently reported by the control panel against
    the docroots stored in the registry.  Orphaned entries (docroots
    no longer valid) are removed and their kernel LVEs destroyed;
    missing entries (current docroots not yet registered) get a fresh
    domain ID assigned and mapped to the user's LVP.
    """
    registry = ws_id_registry.get_all_entries(uid)
    if not registry:
        return

try:
        config = ws_config.LvdConfig.load(uid)
    except (LvdError, OSError) as exc:
        syslog.syslog(syslog.LOG_WARNING,
                      f"regenerate-domains: failed to load LVD config "
                      f"for uid {uid}: {exc}")
        return

current_docroots = set()
    for entry in config.domains:
        try:
            current_docroots.add(ws_config.resolve_docroot(entry.name))
        except LvdError as exc:
            syslog.syslog(syslog.LOG_WARNING,
                          f"regenerate-domains: cannot resolve docroot "
                          f"for domain '{entry.name}': {exc}")

for old_dr, domain_id in list(registry.items()):
        if old_dr in current_docroots:
            continue
        if lve is not None and \
                lve.proc.check_inside_list(domain_id, lvp_id=uid):
            pylve.lve_destroy(domain_id)
        ws_id_registry.remove_domain_id(uid, old_dr)

changed = False
    for dr in current_docroots:
        if dr in registry:
            continue
        domain_id = ws_id_registry.assign_domain_id(uid, dr)
        if lve is not None:
            pylve.lve_lvp_map(uid, domain_id)
        changed = True

if changed and lve is not None:
        domain_map = lve._build_domain_map(uid)
        lve.apply_domain_lve_limits(uid, domain_map=domain_map)

def remove_absent_users():
    """
    Remove from LVE all users, which absent in system
    :return: None
    """
    # Build the set of all known domain LVE IDs from the registry so we can
    # skip them below — they have no system user of their own and their
    # kernel mappings are owned by _sync_map.
    domain_lve_ids: set[int] = set()
    if lve.py.domains_supported():
        domain_lve_ids = ws_id_registry.get_all_domain_ids()

for lve_id, lvp_id in lve.proc.map().items():
        if lve_id in domain_lve_ids:
            continue  # domain LVE — skip, managed by _sync_map
        try:
            pwd.getpwuid(lve_id)  # Check the existence of the user
        except KeyError:
            try:
                lve.lve_destroy(lve_id)  # Destroy lve, if user not exist
            except PyLveError:  # If lve not exist
                try:
                    lve.py.lve_create(lve_id)  # we create lve
                    lve.lve_destroy(lve_id)  # and destroy them again
                    # After destroing lve, mapping will be cleansed of absent lve_id
                except PyLveError:
                    pass

Batal

⬅ python3.11 264 item

Nama	Tipe	Ukuran	Diubah	Perm	Aksi
📁 GitPython-3.1.32.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 Jinja2-3.0.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 Mako-1.2.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 MarkupSafe-2.1.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 PyJWT-2.8.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 PyMySQL-1.1.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 PyVirtualDisplay-3.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 PyYAML-6.0.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 __pycache__	dir	—	23/06/26 03:09	0755	📂
📁 _distutils_hack	dir	—	06/02/26 03:10	0755	📂
📁 _pytest	dir	—	06/02/26 03:10	0755	📂
📁 _yaml	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp-3.9.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp_jinja2	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp_jinja2-1.5.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp_security	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp_security-0.4.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp_session	dir	—	06/02/26 03:10	0755	📂
📁 aiohttp_session-2.9.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 aiosignal	dir	—	06/02/26 03:10	0755	📂
📁 aiosignal-1.3.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 alembic	dir	—	06/02/26 03:10	0755	📂
📁 alembic-1.11.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 astroid	dir	—	06/02/26 03:10	0755	📂
📁 astroid-2.15.6.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 attr	dir	—	06/02/26 03:10	0755	📂
📁 attrs	dir	—	06/02/26 03:10	0755	📂
📁 attrs-23.1.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 backports	dir	—	06/02/26 03:10	0755	📂
📁 certifi	dir	—	06/02/26 03:10	0755	📂
📁 certifi-2023.7.22.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 cffi	dir	—	06/02/26 03:10	0755	📂
📁 cffi-1.15.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 chardet	dir	—	06/02/26 03:10	0755	📂
📁 chardet-5.2.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 charset_normalizer	dir	—	06/02/26 03:10	0755	📂
📁 charset_normalizer-2.1.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 cl_dom_collector	dir	—	23/06/26 03:09	0755	📂
📁 cl_website_collector	dir	—	23/06/26 03:09	0755	📂
📁 clcagefslib	dir	—	23/06/26 03:09	0755	📂
📁 clcommon	dir	—	17/06/26 03:09	0755	📂
📁 clconfig	dir	—	23/06/26 03:09	0755	📂
📁 clconfigure	dir	—	23/06/26 03:09	0755	📂
📁 cldashboard	dir	—	23/06/26 03:09	0755	📂
📁 clevents	dir	—	23/06/26 03:09	0755	📂
📁 clflags	dir	—	17/06/26 03:09	0755	📂
📁 cllicense	dir	—	23/06/26 03:09	0755	📂
📁 cllimits	dir	—	23/06/26 03:09	0755	📂
📁 cllimits_validator	dir	—	23/06/26 03:09	0755	📂
📁 cllimitslib_v2	dir	—	23/06/26 03:09	0755	📂
📁 cllvectl	dir	—	23/06/26 03:09	0755	📂
📁 clpackages	dir	—	23/06/26 03:09	0755	📂
📁 clquota	dir	—	10/07/26 03:10	0755	📂
📁 clselect	dir	—	10/07/26 03:10	0755	📂
📁 clselector	dir	—	10/07/26 03:10	0755	📂
📁 clsentry	dir	—	17/06/26 03:09	0755	📂
📁 clsummary	dir	—	23/06/26 03:09	0755	📂
📁 clveconfig	dir	—	23/06/26 03:09	0755	📂
📁 clwizard	dir	—	23/06/26 03:09	0755	📂
📁 clwpos	dir	—	09/07/26 03:10	0755	📂
📁 configparser-5.0.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 contextlib2	dir	—	06/02/26 03:10	0755	📂
📁 contextlib2-21.6.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 coverage	dir	—	06/02/26 03:10	0755	📂
📁 coverage-7.2.7.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 cryptography	dir	—	06/02/26 03:10	0755	📂
📁 cryptography-41.0.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 ddt-1.4.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 dill	dir	—	06/02/26 03:10	0755	📂
📁 dill-0.3.7.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 distlib	dir	—	06/02/26 03:10	0755	📂
📁 distlib-0.3.8.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 docopt-0.6.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 dodgy	dir	—	06/02/26 03:10	0755	📂
📁 dodgy-0.2.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 filelock	dir	—	06/02/26 03:10	0755	📂
📁 filelock-3.13.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 flake8	dir	—	06/02/26 03:10	0755	📂
📁 flake8-5.0.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 flake8_polyfill	dir	—	06/02/26 03:10	0755	📂
📁 flake8_polyfill-1.0.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 frozenlist	dir	—	06/02/26 03:10	0755	📂
📁 frozenlist-1.4.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 future	dir	—	06/02/26 03:10	0755	📂
📁 future-0.18.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 git	dir	—	06/02/26 03:10	0755	📂
📁 gitdb	dir	—	06/02/26 03:10	0755	📂
📁 gitdb-4.0.10.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 guppy	dir	—	06/02/26 03:10	0755	📂
📁 guppy3-3.1.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 idna	dir	—	06/02/26 03:10	0755	📂
📁 idna-3.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 iniconfig	dir	—	06/02/26 03:10	0755	📂
📁 iniconfig-2.0.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 isort	dir	—	06/02/26 03:10	0755	📂
📁 isort-5.12.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 jinja2	dir	—	06/02/26 03:10	0755	📂
📁 jsonschema	dir	—	06/02/26 03:10	0755	📂
📁 jsonschema-3.2.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 jwt	dir	—	06/02/26 03:10	0755	📂
📁 lazy_object_proxy	dir	—	06/02/26 03:10	0755	📂
📁 lazy_object_proxy-1.9.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 libfuturize	dir	—	06/02/26 03:10	0755	📂
📁 libpasteurize	dir	—	06/02/26 03:10	0755	📂
📁 lve_stats-2.0.dist-info	dir	—	01/07/26 03:10	0755	📂
📁 lve_utils	dir	—	23/06/26 03:09	0755	📂
📁 lvemanager	dir	—	10/07/26 03:10	0755	📂
📁 lvestats	dir	—	01/07/26 03:10	0755	📂
📁 lxml	dir	—	06/02/26 03:10	0755	📂
📁 lxml-4.9.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 mako	dir	—	06/02/26 03:10	0755	📂
📁 markupsafe	dir	—	06/02/26 03:10	0755	📂
📁 mccabe-0.7.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 mock	dir	—	06/02/26 03:10	0755	📂
📁 mock-5.1.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 multidict	dir	—	06/02/26 03:10	0755	📂
📁 multidict-6.0.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 numpy	dir	—	06/02/26 03:10	0755	📂
📁 numpy-1.25.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 numpy.libs	dir	—	06/02/26 03:10	0755	📂
📁 packaging	dir	—	06/02/26 03:10	0755	📂
📁 packaging-23.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 past	dir	—	06/02/26 03:10	0755	📂
📁 pep8_naming-0.10.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pip	dir	—	06/02/26 03:10	0755	📂
📁 pip-25.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pkg_resources	dir	—	06/02/26 03:10	0755	📂
📁 platformdirs	dir	—	06/02/26 03:10	0755	📂
📁 platformdirs-3.11.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pluggy	dir	—	06/02/26 03:10	0755	📂
📁 pluggy-1.2.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 prettytable	dir	—	06/02/26 03:10	0755	📂
📁 prettytable-3.8.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 prometheus_client	dir	—	06/02/26 03:10	0755	📂
📁 prometheus_client-0.8.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 prospector	dir	—	06/02/26 03:10	0755	📂
📁 prospector-1.10.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 psutil	dir	—	06/02/26 03:10	0755	📂
📁 psutil-5.9.5.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 psycopg2	dir	—	06/02/26 03:10	0755	📂
📁 psycopg2_binary-2.9.6.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 psycopg2_binary.libs	dir	—	06/02/26 03:10	0755	📂
📁 pycodestyle-2.9.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pycparser	dir	—	06/02/26 03:10	0755	📂
📁 pycparser-2.21.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pydocstyle	dir	—	06/02/26 03:10	0755	📂
📁 pydocstyle-6.3.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pyfakefs	dir	—	06/02/26 03:12	0755	📂
📁 pyfakefs-5.10.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pyflakes	dir	—	06/02/26 03:10	0755	📂
📁 pyflakes-2.5.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pylint	dir	—	06/02/26 03:10	0755	📂
📁 pylint-2.17.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pylint_celery	dir	—	06/02/26 03:10	0755	📂
📁 pylint_celery-0.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pylint_django	dir	—	06/02/26 03:10	0755	📂
📁 pylint_django-2.5.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pylint_flask	dir	—	06/02/26 03:10	0755	📂
📁 pylint_flask-0.6.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pylint_plugin_utils	dir	—	06/02/26 03:10	0755	📂
📁 pylint_plugin_utils-0.7.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pylve-2.1-py3.11.egg-info	dir	—	02/06/26 03:13	0755	📂
📁 pymysql	dir	—	06/02/26 03:10	0755	📂
📁 pyparsing	dir	—	06/02/26 03:10	0755	📂
📁 pyparsing-3.0.9.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pyrsistent	dir	—	06/02/26 03:10	0755	📂
📁 pyrsistent-0.19.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pytest	dir	—	06/02/26 03:10	0755	📂
📁 pytest-7.4.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pytest_check	dir	—	06/02/26 03:10	0755	📂
📁 pytest_check-2.5.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pytest_snapshot	dir	—	06/02/26 03:10	0755	📂
📁 pytest_snapshot-0.9.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pytest_subprocess	dir	—	06/02/26 03:10	0755	📂
📁 pytest_subprocess-1.5.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pytest_tap	dir	—	06/02/26 03:10	0755	📂
📁 pytest_tap-3.5.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 python_pam-1.8.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 pyvirtualdisplay	dir	—	06/02/26 03:10	0755	📂
📁 raven	dir	—	06/02/26 03:10	0755	📂
📁 raven-6.10.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 requests	dir	—	06/02/26 03:10	0755	📂
📁 requests-2.31.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 requirements_detector	dir	—	06/02/26 03:10	0755	📂
📁 requirements_detector-1.2.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 schema-0.7.5.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 semver	dir	—	06/02/26 03:10	0755	📂
📁 semver-3.0.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 sentry_sdk	dir	—	06/02/26 03:10	0755	📂
📁 sentry_sdk-1.29.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 setoptconf	dir	—	06/02/26 03:10	0755	📂
📁 setoptconf_tmp-0.3.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 setuptools	dir	—	06/02/26 03:12	0755	📂
📁 setuptools-80.9.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 simplejson	dir	—	06/02/26 03:10	0755	📂
📁 simplejson-3.19.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 six-1.16.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 smmap	dir	—	06/02/26 03:10	0755	📂
📁 smmap-5.0.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 snowballstemmer	dir	—	06/02/26 03:10	0755	📂
📁 snowballstemmer-2.2.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 sqlalchemy	dir	—	06/02/26 03:10	0755	📂
📁 sqlalchemy-1.3.24.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 ssa	dir	—	09/07/26 03:09	0755	📂
📁 svgwrite	dir	—	06/02/26 03:10	0755	📂
📁 svgwrite-1.4.3.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 tap	dir	—	06/02/26 03:10	0755	📂
📁 tap_py-3.2.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 testfixtures	dir	—	06/02/26 03:10	0755	📂
📁 testfixtures-7.1.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 toml	dir	—	06/02/26 03:10	0755	📂
📁 toml-0.10.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 tomlkit	dir	—	06/02/26 03:10	0755	📂
📁 tomlkit-0.11.8.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 typing_extensions-4.7.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 unshare-0.22.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 urllib3	dir	—	06/02/26 03:10	0755	📂
📁 urllib3-2.0.4.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 vendors_api	dir	—	17/06/26 03:09	0755	📂
📁 virtualenv	dir	—	06/02/26 03:10	0755	📂
📁 virtualenv-20.21.1.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 wcwidth	dir	—	06/02/26 03:10	0755	📂
📁 wcwidth-0.2.6.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 websiteisolation	dir	—	23/06/26 03:09	0755	📂
📁 wmt	dir	—	09/07/26 03:09	0755	📂
📁 wrapt	dir	—	06/02/26 03:10	0755	📂
📁 wrapt-1.15.0.dist-info	dir	—	06/02/26 03:10	0755	📂
📁 xray	dir	—	09/07/26 03:10	0755	📂
📁 yaml	dir	—	06/02/26 03:10	0755	📂
📁 yarl	dir	—	06/02/26 03:10	0755	📂
📁 yarl-1.9.2.dist-info	dir	—	06/02/26 03:10	0755	📂
📄 _cffi_backend.cpython-311-x86_64-linux-gnu.so	so	267.6 KB	20/01/26 19:58	0755	✏️ ⬇
📄 _lvdmap.cpython-311-x86_64-linux-gnu.so	so	18.4 KB	03/06/26 19:06	0755	✏️ ⬇
🐍 _pyrsistent_version.py	py	23 B	20/01/26 19:58	0644	✏️ ⬇
🐍 cl_proc_hidepid.py	py	4.5 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 clcontrollib.py	py	51.8 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 cldetectlib.py	py	18.4 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 cldiaglib.py	py	48.5 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 clhooklib.py	py	1.3 KB	01/07/26 16:05	0644	✏️ ⬇
🐍 cli_utils.py	py	1.7 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 cllicenselib.py	py	9.1 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 clsetuplib.py	py	4.3 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 clsudo.py	py	16.7 KB	03/06/26 20:44	0644	✏️ ⬇
🐍 configparser.py	py	1.5 KB	20/01/26 19:58	0644	✏️ ⬇
🐍 ddt.py	py	12.4 KB	20/01/26 19:58	0644	✏️ ⬇
📄 distutils-precedence.pth	pth	151 B	20/01/26 19:58	0644	✏️ ⬇
🐍 docopt.py	py	19.5 KB	20/01/26 19:58	0644	✏️ ⬇
🐍 lveapi.py	py	19.6 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 lvectllib.py	py	125.9 KB	03/06/26 17:46	0644	✏️ ⬇
🐍 lvestat.py	py	6.8 KB	03/06/26 20:44	0644	✏️ ⬇
🐍 mccabe.py	py	10.4 KB	20/01/26 19:58	0644	✏️ ⬇
🐍 pam.py	py	7.4 KB	20/01/26 19:58	0644	✏️ ⬇
🐍 pep8ext_naming.py	py	18.6 KB	20/01/26 19:58	0644	✏️ ⬇
🐍 py.py	py	263 B	20/01/26 19:58	0644	✏️ ⬇
🐍 pycodestyle.py	py	101.1 KB	20/01/26 19:58	0644	✏️ ⬇
📄 pylve.cpython-311-x86_64-linux-gnu.so	so	28.8 KB	18/05/26 20:19	0755	✏️ ⬇
🐍 remove_ubc.py	py	5.7 KB	03/06/26 17:46	0755	✏️ ⬇
🐍 schema.py	py	29.5 KB	20/01/26 19:58	0644	✏️ ⬇
🐍 secureio.py	py	19.1 KB	03/06/26 20:44	0644	✏️ ⬇
📄 simple_rpm.so	so	15.2 KB	29/05/26 16:34	0755	✏️ ⬇
🐍 six.py	py	33.7 KB	20/01/26 19:58	0644	✏️ ⬇
🐍 typing_extensions.py	py	108.5 KB	20/01/26 19:58	0644	✏️ ⬇
📄 unshare.cpython-311-x86_64-linux-gnu.so	so	16.3 KB	20/01/26 19:58	0755	✏️ ⬇

✏️ Rename

🔑 Ubah Permission