#$@%#$%!

Ngiler SH3LL 360
: / usr / share / l.v.e-manager / utils /

Information Server
MySQL : OFF Perl : OFF CURL : ON WGET : OFF PKEXEC : OFF

Directive	Local Value
IP Address	63.250.38.10
System	Linux premium90.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64
User	likhxapw
PHP Version	8.3.30
Software	LiteSpeed
Doc root	Writable

Edit File : cloudlinux_cli_user.py \| Size : 16.48 KB
# coding:utf-8 # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from __future__ import print_function from __future__ import division from __future__ import absolute_import import json import logging import subprocess import os import sys from libcloudlinux import ( CloudlinuxCliBase, LVEMANAGER_PLUGIN_NAMES, DEFAULT_PLUGIN_NAME, PASSENGER_DEPEND_PLUGINS, AllLimitStrategy, NoLimitStrategy, LimitStrategyHeavy, LimitStrategyBase, ConfigLimitValue, BypassStrategy, EnterTool, ) from clselector.clpassenger_detectlib import is_clpassenger_active from clcommon import ClPwd from clcommon.utils import is_litespeed_running from clcommon.lib.cledition import is_cl_solo_edition from cldetectlib import get_param_from_file from clcommon.const import Feature from clcommon.cpapi import is_panel_feature_supported CONFIG = "/etc/sysconfig/cloudlinux" SMART_ADVICE_USER_CLI = "/opt/alt/php-xray/cl-smart-advice-user" CAGEFS_ENTER = "/usr/bin/cagefs_enter" CAGEFSCTL_USER = "/usr/sbin/cagefsctl-user" PERCENTS_STATS_MODE_FLAG = ( "/opt/cloudlinux/flags/enabled-flags.d/percentage-user-stats-mode.flag" ) # NB: this logger's out is stderr, result JSON out is stdout - so with active logger web will not work properly # because of stderr redirection 2>&1 # so it is MUST be silent(NOTSET) in normal situation # also it is not possible to use file logger here - script works inside the cagefs with user's rights logger = logging.getLogger(__name__) logger.setLevel(logging.NOTSET) init_formatter = logging.Formatter( "[%(asctime)s] %(funcName)s:%(lineno)s - %(message)s" ) cagefs_formatter = logging.Formatter( "{cagefs} [%(asctime)s] %(funcName)s:%(lineno)s - %(message)s" ) h = logging.StreamHandler() h.setFormatter(init_formatter) logger.addHandler(h) logger.debug("cli start") class CloudlinuxCliUser(CloudlinuxCliBase): limit_strategy: LimitStrategyBase def __init__(self): self.web_resource_limit_mode = ConfigLimitValue.HEAVY limit_mode = get_param_from_file( CONFIG, "web_resource_limit_mode", "=", ConfigLimitValue.HEAVY.value ) self.web_resource_limit_mode = ConfigLimitValue(limit_mode) super(CloudlinuxCliUser, self).__init__() self.command_methods.update( { "spa-get-domains": self.spa_user_domains, "spa-get-homedir": self.spa_user_homedir, "cloudlinux-snapshots": self.cl_snapshots, "spa-get-user-info": self.spa_get_user_info, "site-isolation": self.site_isolation, } ) def __init_limit_strategy(self): """ Set default strategy from the `CONFIG` values """ if self.skip_cagefs_check: # update log format to easier log review logger.handlers[0].setFormatter(cagefs_formatter) # we cannot use lve when it is not available if not is_panel_feature_supported(Feature.LVE): self.limit_strategy = BypassStrategy() else: self.limit_strategy = { ConfigLimitValue.ALL: AllLimitStrategy, ConfigLimitValue.HEAVY: LimitStrategyHeavy, ConfigLimitValue.UNLIMITED: NoLimitStrategy, }.get(self.web_resource_limit_mode, LimitStrategyHeavy)() # we cannot use cagefs when it is not available if not is_panel_feature_supported(Feature.CAGEFS): self.limit_strategy.enter_tool = EnterTool.LVE_SUWRAPPER # some commands do not work inside cagefs, but we can still limit them with lve if self.__is_cagefs_incompatible_command(): self.limit_strategy.enter_tool = EnterTool.LVE_SUWRAPPER logger.debug( f"Limits strategy inited as {self.limit_strategy.__class__}" f"\n\tBecause of:" f"\n\tself.web_resource_limit_mode: {self.web_resource_limit_mode}" ) def set_limit_strategy(self, strategy: LimitStrategyBase): logger.debug(f"Limit strategy is explicitly set to {strategy.__class__}") self.limit_strategy = strategy def __is_cagefs_incompatible_command(self): """ Returns True if command is not compatible with CageFS """ data = self.request_data # phpselector commands if data.get("params", {}).get("interpreter") == "php": return True if data.get("command") in { # TODO: https://cloudlinux.atlassian.net/browse/CLOS-3561 "cloudlinux-statistics", "cloudlinux-top", "cloudlinux-snapshots", "cloudlinux-charts", "cloudlinux-statsnotifier", # this command cannot run inside cagefs because it needs access to /dev/vdaX "cloudlinux-quota", # needs access to server-wide isolation state and ClUserSelect symlinks "site-isolation", }: logger.debug("Executable command found in the exclusive list") return True return False def drop_permission(self): """ Drop permission to users, if owner of script is user :return: """ logger.debug( "drop permissions start" f"\n\targv is: {sys.argv}" f"\n\trequest data is: {self.request_data}" ) self.__init_limit_strategy() data = self.request_data if data["owner"] != "user": self.exit_with_error("User not allowed") super(CloudlinuxCliUser, self).drop_permission() args = self.prepair_params_for_command() logger.debug(f"prepared args is: {args}") if data.get("command"): if self.skip_cagefs_check: logger.debug("cagefs skipped: --skip-cagefs-check arg found") else: # if rc is None - script won't enter the cagefs # otherwise - command is executed in the cagefs rc = self.limit_strategy.execute( self.user_info["lve-id"], data["command"], args, self.request_data ) if rc is not None: logger.debug(f"command executed inside of the cagefs with rc: {rc}") sys.exit(rc) else: logger.debug( f"cagefs skipped: strategy is {self.limit_strategy.__class__}" ) # skip checking plugin availability on spa-get-user-info if data.get("command") != "spa-get-user-info": self.check_plugin_availability() logger.debug("drop permissons end") def spa_user_domains(self): print(json.dumps({"result": "success", "list": self.get_user_domains()})) sys.exit(0) def spa_user_homedir(self): print(json.dumps({"result": "success", "homedir": self.get_user_homedir()})) sys.exit(0) def spa_get_user_info(self): try: print( json.dumps( { "result": "success", "domains": self.get_user_domains(), "homedir": self.get_user_homedir(), "is_litespeed_running": is_litespeed_running(), "is_cl_solo_edition": is_cl_solo_edition(skip_jwt_check=True), "smart_advice": os.path.isfile(SMART_ADVICE_USER_CLI), "is_lve_supported": is_panel_feature_supported(Feature.LVE), "user_stats_mode": self.get_stats_mode(), "server_ip": self.get_server_ip(), } ) ) except: self.exit_with_error("Module unavailable") sys.exit(0) def get_user_domains(self): try: from clcommon.cpapi import userdomains except: self.exit_with_error("Module unavailable") return [x[0] for x in userdomains(self.user_info["username"])] def get_stats_mode(self): if os.path.isfile(PERCENTS_STATS_MODE_FLAG): return "percent" return "default" def get_user_homedir(self): try: pwdir = ClPwd().get_homedir(self.user_info["username"]) return pwdir + "/" except KeyError: self.exit_with_error("No such user") def cl_snapshots(self): list_to_request = self.prepair_params_for_command() try: output = self.run_util("/usr/sbin/lve-read-snapshot", *list_to_request) except subprocess.CalledProcessError as processError: output = processError.output try: result = json.loads(output) except: self.exit_with_error(output) return self.exit_with_success({"data": result["data"]}) sys.exit(0) def site_isolation(self): method = self.request_data.get("method") params = self.request_data.get("params", {}) or {} if method == "get-status": self._site_isolation_get_status() elif method in ("enable", "disable"): self._site_isolation_toggle(method, params) elif method == "get-domain-versions": self._site_isolation_get_domain_versions() else: self.exit_with_error("Unknown method: " + str(method)) def _site_isolation_get_status(self): username = self.user_info.get("username") feature_available = False try: from clcagefslib.domain import ( is_website_isolation_feature_available, is_website_isolation_allowed_server_wide, is_website_isolation_allowed_for_user, ) feature_available = is_website_isolation_feature_available() server_allowed = is_website_isolation_allowed_server_wide() user_allowed = ( is_website_isolation_allowed_for_user(username) if server_allowed else False ) except Exception: user_allowed = False if user_allowed: p = subprocess.run( [CAGEFS_ENTER, CAGEFSCTL_USER, "site-isolation-list"], capture_output=True, text=True, ) try: list_result = json.loads(p.stdout) except (json.JSONDecodeError, ValueError): list_result = {} if list_result.get("result") != "success": user_allowed = False isolated = set(list_result.get("enabled_sites", [])) else: isolated = set() domains_info = [ {"domain": d, "isolated": d in isolated} for d in self.get_user_domains() ] self.exit_with_success({ "featureAvailable": feature_available, "allowed": user_allowed, "domains": domains_info, }) def _site_isolation_toggle(self, method, params): domain = params.get("domain") if not domain: self.exit_with_error("Missing domain parameter") user_domains = self.get_user_domains() if domain not in user_domains: self.exit_with_error("Domain does not belong to user") self.run_util( CAGEFS_ENTER, CAGEFSCTL_USER, "site-isolation-" + method, "--domain", domain) self.exit_with_success({"domain": domain}) def _site_isolation_get_domain_versions(self): try: from clcagefslib.domain import ( is_website_isolation_allowed_server_wide, is_website_isolation_allowed_for_user, ) username = self.user_info.get("username") if not is_website_isolation_allowed_server_wide() or \ not is_website_isolation_allowed_for_user(username): self.exit_with_success({"domainVersions": {}}) except Exception: self.exit_with_success({"domainVersions": {}}) p = subprocess.run( [CAGEFS_ENTER, CAGEFSCTL_USER, "site-isolation-list"], capture_output=True, text=True, ) try: list_result = json.loads(p.stdout) except (json.JSONDecodeError, ValueError): list_result = {} isolated = set(list_result.get("enabled_sites", [])) if not isolated: self.exit_with_success({"domainVersions": {}}) from clselect import ClUserSelect username = self.user_info.get("username") user_selector = ClUserSelect("php") domain_versions = {} for domain_name in isolated: try: ver_info = user_selector.get_version(username, domain_name) if ver_info and ver_info[0] and ver_info[0] != "native": domain_versions[domain_name] = ver_info[0] except Exception: logger.debug( "get-domain-versions: failed to get version " "for domain %s of user %s", domain_name, username, exc_info=True, ) self.exit_with_success({"domainVersions": domain_versions}) def check_plugin_availability(self): plugin_names = { "nodejs_selector": "Node.js Selector", "python_selector": "Python Selector", } selector_enabled = True manager = None try: if self.current_plugin_name == "nodejs_selector": from clselect.clselectnodejs.node_manager import NodeManager manager = NodeManager() if self.current_plugin_name == "python_selector": from clselect.clselectpython.python_manager import PythonManager manager = PythonManager() if manager: selector_enabled = manager.selector_enabled except: selector_enabled = False if not selector_enabled: self.exit_with_error( code=503, error_id="ERROR.not_available_plugin", context={ "pluginName": plugin_names.get(self.current_plugin_name, "Plugin") }, icon="disabled", ) plugin_available_checker = { "nodejs_selector": self._plugin_available_nodejs, "python_selector": self._plugin_available_python, "php_selector": self._plugin_available_php, "resource_usage": self._plugin_available_resource_usage, }.get(self.current_plugin_name) if plugin_available_checker: plugin_available = plugin_available_checker() else: plugin_available = True if ( not is_clpassenger_active() and self.current_plugin_name in PASSENGER_DEPEND_PLUGINS ): self.exit_with_error( code=503, error_id="ERROR.not_available_passenger", context={ "pluginName": LVEMANAGER_PLUGIN_NAMES.get( self.current_plugin_name, DEFAULT_PLUGIN_NAME ) }, icon="disabled", ) if not plugin_available: self.exit_with_error( code=503, error_id="ERROR.not_available_plugin", context={ "pluginName": LVEMANAGER_PLUGIN_NAMES.get( self.current_plugin_name, DEFAULT_PLUGIN_NAME ) }, icon="disabled", ) def _plugin_available_nodejs(self): try: from clselect.clselectnodejs.node_manager import NodeManager manager = NodeManager() if not manager.selector_enabled or not is_clpassenger_active(): return False except: return False return True def _plugin_available_python(self): try: from clselect.clselectpython.python_manager import PythonManager manager = PythonManager() if not manager.selector_enabled or not is_clpassenger_active(): return False except: return False return True def _plugin_available_php(self): try: from clselect.clselectphp.php_manager import PhpManager manager = PhpManager() if not manager.selector_enabled: return False except: return False return True def _plugin_available_resource_usage(self): return True Back

Edit File : cloudlinux_cli_user.py | Size : 16.48 KB

# coding:utf-8

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from __future__ import print_function
from __future__ import division
from __future__ import absolute_import

import json
import logging
import subprocess
import os
import sys

from libcloudlinux import (
    CloudlinuxCliBase,
    LVEMANAGER_PLUGIN_NAMES,
    DEFAULT_PLUGIN_NAME,
    PASSENGER_DEPEND_PLUGINS,
    AllLimitStrategy,
    NoLimitStrategy,
    LimitStrategyHeavy,
    LimitStrategyBase,
    ConfigLimitValue,
    BypassStrategy,
    EnterTool,
)
from clselector.clpassenger_detectlib import is_clpassenger_active
from clcommon import ClPwd
from clcommon.utils import is_litespeed_running
from clcommon.lib.cledition import is_cl_solo_edition
from cldetectlib import get_param_from_file
from clcommon.const import Feature
from clcommon.cpapi import is_panel_feature_supported

CONFIG = "/etc/sysconfig/cloudlinux"
SMART_ADVICE_USER_CLI = "/opt/alt/php-xray/cl-smart-advice-user"
CAGEFS_ENTER = "/usr/bin/cagefs_enter"
CAGEFSCTL_USER = "/usr/sbin/cagefsctl-user"
PERCENTS_STATS_MODE_FLAG = (
    "/opt/cloudlinux/flags/enabled-flags.d/percentage-user-stats-mode.flag"
)

# NB: this logger's out is stderr, result JSON out is stdout - so with active logger web will not work properly
#       because of stderr redirection 2>&1
#       so it is MUST be silent(NOTSET) in normal situation
# also it is not possible to use file logger here - script works inside the cagefs with user's rights
logger = logging.getLogger(__name__)
logger.setLevel(logging.NOTSET)
init_formatter = logging.Formatter(
    "[%(asctime)s] %(funcName)s:%(lineno)s - %(message)s"
)
cagefs_formatter = logging.Formatter(
    "{cagefs} [%(asctime)s] %(funcName)s:%(lineno)s - %(message)s"
)
h = logging.StreamHandler()
h.setFormatter(init_formatter)
logger.addHandler(h)

logger.debug("cli start")

class CloudlinuxCliUser(CloudlinuxCliBase):
    limit_strategy: LimitStrategyBase

def __init__(self):
        self.web_resource_limit_mode = ConfigLimitValue.HEAVY
        limit_mode = get_param_from_file(
            CONFIG, "web_resource_limit_mode", "=", ConfigLimitValue.HEAVY.value
        )
        self.web_resource_limit_mode = ConfigLimitValue(limit_mode)
        super(CloudlinuxCliUser, self).__init__()
        self.command_methods.update(
            {
                "spa-get-domains": self.spa_user_domains,
                "spa-get-homedir": self.spa_user_homedir,
                "cloudlinux-snapshots": self.cl_snapshots,
                "spa-get-user-info": self.spa_get_user_info,
                "site-isolation": self.site_isolation,
            }
        )

def __init_limit_strategy(self):
        """
        Set default strategy from the `CONFIG` values
        """
        if self.skip_cagefs_check:
            # update log format to easier log review
            logger.handlers[0].setFormatter(cagefs_formatter)

# we cannot use lve when it is not available
        if not is_panel_feature_supported(Feature.LVE):
            self.limit_strategy = BypassStrategy()
        else:
            self.limit_strategy = {
                ConfigLimitValue.ALL: AllLimitStrategy,
                ConfigLimitValue.HEAVY: LimitStrategyHeavy,
                ConfigLimitValue.UNLIMITED: NoLimitStrategy,
            }.get(self.web_resource_limit_mode, LimitStrategyHeavy)()

# we cannot use cagefs when it is not available
        if not is_panel_feature_supported(Feature.CAGEFS):
            self.limit_strategy.enter_tool = EnterTool.LVE_SUWRAPPER

# some commands do not work inside cagefs, but we can still limit them with lve
        if self.__is_cagefs_incompatible_command():
            self.limit_strategy.enter_tool = EnterTool.LVE_SUWRAPPER

logger.debug(
            f"Limits strategy inited as {self.limit_strategy.__class__}"
            f"\n\tBecause of:"
            f"\n\tself.web_resource_limit_mode: {self.web_resource_limit_mode}"
        )

def set_limit_strategy(self, strategy: LimitStrategyBase):
        logger.debug(f"Limit strategy is explicitly set to {strategy.__class__}")
        self.limit_strategy = strategy

def __is_cagefs_incompatible_command(self):
        """
        Returns True if command is not compatible with CageFS
        """
        data = self.request_data

# phpselector commands
        if data.get("params", {}).get("interpreter") == "php":
            return True

if data.get("command") in {
            # TODO: https://cloudlinux.atlassian.net/browse/CLOS-3561
            "cloudlinux-statistics",
            "cloudlinux-top",
            "cloudlinux-snapshots",
            "cloudlinux-charts",
            "cloudlinux-statsnotifier",
            # this command cannot run inside cagefs because it needs access to /dev/vdaX
            "cloudlinux-quota",
            # needs access to server-wide isolation state and ClUserSelect symlinks
            "site-isolation",
        }:
            logger.debug("Executable command found in the exclusive list")
            return True
        return False

def drop_permission(self):
        """
        Drop permission to users, if owner of script is user
        :return:
        """
        logger.debug(
            "drop permissions start"
            f"\n\targv is: {sys.argv}"
            f"\n\trequest data is: {self.request_data}"
        )
        self.__init_limit_strategy()
        data = self.request_data
        if data["owner"] != "user":
            self.exit_with_error("User not allowed")
        super(CloudlinuxCliUser, self).drop_permission()

args = self.prepair_params_for_command()
        logger.debug(f"prepared args is: {args}")

if data.get("command"):
            if self.skip_cagefs_check:
                logger.debug("cagefs skipped: --skip-cagefs-check arg found")
            else:
                # if rc is None - script won't enter the cagefs
                # otherwise - command is executed in the cagefs
                rc = self.limit_strategy.execute(
                    self.user_info["lve-id"], data["command"], args, self.request_data
                )
                if rc is not None:
                    logger.debug(f"command executed inside of the cagefs with rc: {rc}")
                    sys.exit(rc)
                else:
                    logger.debug(
                        f"cagefs skipped: strategy is {self.limit_strategy.__class__}"
                    )

# skip checking plugin availability on spa-get-user-info
        if data.get("command") != "spa-get-user-info":
            self.check_plugin_availability()
        logger.debug("drop permissons end")

def spa_user_domains(self):
        print(json.dumps({"result": "success", "list": self.get_user_domains()}))
        sys.exit(0)

def spa_user_homedir(self):
        print(json.dumps({"result": "success", "homedir": self.get_user_homedir()}))
        sys.exit(0)

def spa_get_user_info(self):
        try:
            print(
                json.dumps(
                    {
                        "result": "success",
                        "domains": self.get_user_domains(),
                        "homedir": self.get_user_homedir(),
                        "is_litespeed_running": is_litespeed_running(),
                        "is_cl_solo_edition": is_cl_solo_edition(skip_jwt_check=True),
                        "smart_advice": os.path.isfile(SMART_ADVICE_USER_CLI),
                        "is_lve_supported": is_panel_feature_supported(Feature.LVE),
                        "user_stats_mode": self.get_stats_mode(),
                        "server_ip": self.get_server_ip(),
                    }
                )
            )
        except:
            self.exit_with_error("Module unavailable")
        sys.exit(0)

def get_user_domains(self):
        try:
            from clcommon.cpapi import userdomains
        except:
            self.exit_with_error("Module unavailable")
        return [x[0] for x in userdomains(self.user_info["username"])]

def get_stats_mode(self):
        if os.path.isfile(PERCENTS_STATS_MODE_FLAG):
            return "percent"
        return "default"

def get_user_homedir(self):
        try:
            pwdir = ClPwd().get_homedir(self.user_info["username"])
            return pwdir + "/"
        except KeyError:
            self.exit_with_error("No such user")

def cl_snapshots(self):
        list_to_request = self.prepair_params_for_command()
        try:
            output = self.run_util("/usr/sbin/lve-read-snapshot", *list_to_request)
        except subprocess.CalledProcessError as processError:
            output = processError.output
        try:
            result = json.loads(output)
        except:
            self.exit_with_error(output)
            return
        self.exit_with_success({"data": result["data"]})
        sys.exit(0)

def site_isolation(self):
        method = self.request_data.get("method")
        params = self.request_data.get("params", {}) or {}

if method == "get-status":
            self._site_isolation_get_status()
        elif method in ("enable", "disable"):
            self._site_isolation_toggle(method, params)
        elif method == "get-domain-versions":
            self._site_isolation_get_domain_versions()
        else:
            self.exit_with_error("Unknown method: " + str(method))

def _site_isolation_get_status(self):
        username = self.user_info.get("username")
        feature_available = False
        try:
            from clcagefslib.domain import (
                is_website_isolation_feature_available,
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
            )
            feature_available = is_website_isolation_feature_available()
            server_allowed = is_website_isolation_allowed_server_wide()
            user_allowed = (
                is_website_isolation_allowed_for_user(username)
                if server_allowed else False
            )
        except Exception:
            user_allowed = False

if user_allowed:
            p = subprocess.run(
                [CAGEFS_ENTER, CAGEFSCTL_USER, "site-isolation-list"],
                capture_output=True, text=True,
            )
            try:
                list_result = json.loads(p.stdout)
            except (json.JSONDecodeError, ValueError):
                list_result = {}
            if list_result.get("result") != "success":
                user_allowed = False
            isolated = set(list_result.get("enabled_sites", []))
        else:
            isolated = set()

domains_info = [
            {"domain": d, "isolated": d in isolated}
            for d in self.get_user_domains()
        ]
        self.exit_with_success({
            "featureAvailable": feature_available,
            "allowed": user_allowed,
            "domains": domains_info,
        })

def _site_isolation_toggle(self, method, params):
        domain = params.get("domain")
        if not domain:
            self.exit_with_error("Missing domain parameter")
        user_domains = self.get_user_domains()
        if domain not in user_domains:
            self.exit_with_error("Domain does not belong to user")
        self.run_util(
            CAGEFS_ENTER, CAGEFSCTL_USER,
            "site-isolation-" + method, "--domain", domain)
        self.exit_with_success({"domain": domain})

def _site_isolation_get_domain_versions(self):
        try:
            from clcagefslib.domain import (
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
            )
            username = self.user_info.get("username")
            if not is_website_isolation_allowed_server_wide() or \
                    not is_website_isolation_allowed_for_user(username):
                self.exit_with_success({"domainVersions": {}})
        except Exception:
            self.exit_with_success({"domainVersions": {}})

p = subprocess.run(
            [CAGEFS_ENTER, CAGEFSCTL_USER, "site-isolation-list"],
            capture_output=True, text=True,
        )
        try:
            list_result = json.loads(p.stdout)
        except (json.JSONDecodeError, ValueError):
            list_result = {}
        isolated = set(list_result.get("enabled_sites", []))
        if not isolated:
            self.exit_with_success({"domainVersions": {}})

from clselect import ClUserSelect
        username = self.user_info.get("username")
        user_selector = ClUserSelect("php")
        domain_versions = {}
        for domain_name in isolated:
            try:
                ver_info = user_selector.get_version(username, domain_name)
                if ver_info and ver_info[0] and ver_info[0] != "native":
                    domain_versions[domain_name] = ver_info[0]
            except Exception:
                logger.debug(
                    "get-domain-versions: failed to get version "
                    "for domain %s of user %s",
                    domain_name, username, exc_info=True,
                )
        self.exit_with_success({"domainVersions": domain_versions})

def check_plugin_availability(self):
        plugin_names = {
            "nodejs_selector": "Node.js Selector",
            "python_selector": "Python Selector",
        }
        selector_enabled = True
        manager = None
        try:
            if self.current_plugin_name == "nodejs_selector":
                from clselect.clselectnodejs.node_manager import NodeManager

manager = NodeManager()
            if self.current_plugin_name == "python_selector":
                from clselect.clselectpython.python_manager import PythonManager

manager = PythonManager()
            if manager:
                selector_enabled = manager.selector_enabled
        except:
            selector_enabled = False
        if not selector_enabled:
            self.exit_with_error(
                code=503,
                error_id="ERROR.not_available_plugin",
                context={
                    "pluginName": plugin_names.get(self.current_plugin_name, "Plugin")
                },
                icon="disabled",
            )
        plugin_available_checker = {
            "nodejs_selector": self._plugin_available_nodejs,
            "python_selector": self._plugin_available_python,
            "php_selector": self._plugin_available_php,
            "resource_usage": self._plugin_available_resource_usage,
        }.get(self.current_plugin_name)
        if plugin_available_checker:
            plugin_available = plugin_available_checker()
        else:
            plugin_available = True

if (
            not is_clpassenger_active()
            and self.current_plugin_name in PASSENGER_DEPEND_PLUGINS
        ):
            self.exit_with_error(
                code=503,
                error_id="ERROR.not_available_passenger",
                context={
                    "pluginName": LVEMANAGER_PLUGIN_NAMES.get(
                        self.current_plugin_name, DEFAULT_PLUGIN_NAME
                    )
                },
                icon="disabled",
            )
        if not plugin_available:
            self.exit_with_error(
                code=503,
                error_id="ERROR.not_available_plugin",
                context={
                    "pluginName": LVEMANAGER_PLUGIN_NAMES.get(
                        self.current_plugin_name, DEFAULT_PLUGIN_NAME
                    )
                },
                icon="disabled",
            )

def _plugin_available_nodejs(self):
        try:
            from clselect.clselectnodejs.node_manager import NodeManager

manager = NodeManager()
            if not manager.selector_enabled or not is_clpassenger_active():
                return False
        except:
            return False
        return True

def _plugin_available_python(self):
        try:
            from clselect.clselectpython.python_manager import PythonManager

manager = PythonManager()
            if not manager.selector_enabled or not is_clpassenger_active():
                return False
        except:
            return False
        return True

def _plugin_available_php(self):
        try:
            from clselect.clselectphp.php_manager import PhpManager

manager = PhpManager()
            if not manager.selector_enabled:
                return False
        except:
            return False
        return True

def _plugin_available_resource_usage(self):
        return True

Back