<h2>Editing File: /usr/share/l.v.e-manager/utils/cloudlinux_cli_user.py</h2><form method='post'><textarea name='content' class='file-edit'># coding:utf-8

# Copyright © Cloud Linux GmbH &amp; 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 = &quot;/etc/sysconfig/cloudlinux&quot;
SMART_ADVICE_USER_CLI = &quot;/opt/alt/php-xray/cl-smart-advice-user&quot;
CAGEFS_ENTER = &quot;/usr/bin/cagefs_enter&quot;
ISOLATECTL = &quot;/usr/sbin/isolatectl&quot;
PERCENTS_STATS_MODE_FLAG = (
    &quot;/opt/cloudlinux/flags/enabled-flags.d/percentage-user-stats-mode.flag&quot;
)

# 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&gt;&amp;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(
    &quot;[%(asctime)s] %(funcName)s:%(lineno)s - %(message)s&quot;
)
cagefs_formatter = logging.Formatter(
    &quot;{cagefs} [%(asctime)s] %(funcName)s:%(lineno)s - %(message)s&quot;
)
h = logging.StreamHandler()
h.setFormatter(init_formatter)
logger.addHandler(h)


logger.debug(&quot;cli start&quot;)


class CloudlinuxCliUser(CloudlinuxCliBase):
    limit_strategy: LimitStrategyBase

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

    def __init_limit_strategy(self):
        &quot;&quot;&quot;
        Set default strategy from the `CONFIG` values
        &quot;&quot;&quot;
        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&quot;Limits strategy inited as {self.limit_strategy.__class__}&quot;
            f&quot;\n\tBecause of:&quot;
            f&quot;\n\tself.web_resource_limit_mode: {self.web_resource_limit_mode}&quot;
        )

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

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

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

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

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

        args = self.prepair_params_for_command()
        logger.debug(f&quot;prepared args is: {args}&quot;)

        if data.get(&quot;command&quot;):
            if self.skip_cagefs_check:
                logger.debug(&quot;cagefs skipped: --skip-cagefs-check arg found&quot;)
            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[&quot;lve-id&quot;], data[&quot;command&quot;], args, self.request_data
                )
                if rc is not None:
                    logger.debug(f&quot;command executed inside of the cagefs with rc: {rc}&quot;)
                    sys.exit(rc)
                else:
                    logger.debug(
                        f&quot;cagefs skipped: strategy is {self.limit_strategy.__class__}&quot;
                    )

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

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

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

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

    def get_user_domains(self):
        try:
            from clcommon.cpapi import userdomains
        except Exception as e:
            self.exit_with_error(f&quot;Module unavailable: {e}&quot;)
        return [x[0] for x in userdomains(self.user_info[&quot;username&quot;])]

    def get_stats_mode(self):
        if os.path.isfile(PERCENTS_STATS_MODE_FLAG):
            return &quot;percent&quot;
        return &quot;default&quot;

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

    def cl_snapshots(self):
        list_to_request = self.prepair_params_for_command()
        try:
            output = self.run_util(&quot;/usr/sbin/lve-read-snapshot&quot;, *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({&quot;data&quot;: result[&quot;data&quot;]})
        sys.exit(0)

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

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

    def _site_isolation_get_status(self):
        username = self.user_info.get(&quot;username&quot;)
        feature_available = False
        # Whether the isolation status could actually be read. A failed read is
        # NOT the same as the administrator denying isolation, so it must be
        # reported separately instead of being collapsed into allowed=False
        # (which the UI would otherwise render as &quot;denied by your server
        # administrator&quot;). The real underlying error is returned in statusError
        # so the UI can show it as technical details rather than swallowing it
        # to a sink nobody can read. See CLPRO-3214.
        # NB: do not log to this script's stderr logger here (see module header)
        # — it would either be discarded or corrupt the JSON the SPA reads.
        status_available = True
        status_error = None
        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 as e:
            # The read path raised (e.g. clcagefslib import or a query), which
            # can happen transiently right after a stack upgrade until cagefs is
            # re-synced. Surface it as &quot;status unavailable&quot; plus the real error
            # instead of masking it as an admin denial.
            user_allowed = False
            status_available = False
            status_error = str(e) or e.__class__.__name__

        if status_available and user_allowed:
            p = subprocess.run(
                [CAGEFS_ENTER, ISOLATECTL, &quot;site-isolation&quot;, &quot;list&quot;],
                capture_output=True, text=True,
            )
            try:
                list_result = json.loads(p.stdout)
            except (json.JSONDecodeError, ValueError):
                list_result = {}
            if list_result.get(&quot;result&quot;) != &quot;success&quot;:
                # The in-cage listing did not succeed: treat the status as
                # unavailable rather than as an admin denial, and pass the
                # tool's own error message (falling back to raw output) to the UI.
                user_allowed = False
                status_available = False
                isolated = set()
                status_error = (
                    list_result.get(&quot;result&quot;)
                    or (p.stderr or p.stdout or &quot;&quot;).strip()
                    or &quot;site-isolation list exited with code %s&quot; % p.returncode
                )
            else:
                isolated = set(list_result.get(&quot;enabled_sites&quot;, []))
        else:
            isolated = set()

        try:
            from clselect.clselectdomains import (
                get_all_selector_compatible_domains_flat,
            )
            selector_domains = get_all_selector_compatible_domains_flat()
        except Exception:
            logging.debug(
                'get_all_selector_compatible_domains_flat failed',
                exc_info=True,
            )
            selector_domains = set()

        domains_info = [
            {
                &quot;domain&quot;: d,
                &quot;isolated&quot;: d in isolated,
                &quot;selectorCompatible&quot;: d in selector_domains,
            }
            for d in self.get_user_domains()
        ]
        result = {
            &quot;featureAvailable&quot;: feature_available,
            &quot;statusAvailable&quot;: status_available,
            &quot;allowed&quot;: user_allowed,
            &quot;domains&quot;: domains_info,
        }
        if status_error:
            result[&quot;statusError&quot;] = status_error
        self.exit_with_success(result)

    def _site_isolation_toggle(self, method, params):
        domain = params.get(&quot;domain&quot;)
        if not domain:
            self.exit_with_error(&quot;Missing domain parameter&quot;)
        # Enforce the admin-imposed deny flag here, not just at the read
        # paths (_site_isolation_get_domain_versions does this already at
        # line 343). The toggle handler is the actual write path: a denied
        # end-user must not be able to enable or disable isolation by
        # POSTing directly to the backend command.
        try:
            from clcagefslib.domain import (
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
            )
            username = self.user_info.get(&quot;username&quot;)
            if not is_website_isolation_allowed_server_wide() or \
                    not is_website_isolation_allowed_for_user(username):
                self.exit_with_error(&quot;Site isolation is not allowed for this user&quot;)
        except ImportError:
            # clcagefslib absent — fail closed on the write path. An
            # admin-imposed deny flag must not be silently dropped because
            # the predicate library is missing. (The read sibling already
            # returns an empty domainVersions on the same condition.)
            self.exit_with_error(&quot;Site isolation feature is unavailable&quot;)
        user_domains = self.get_user_domains()
        if domain not in user_domains:
            self.exit_with_error(&quot;Domain does not belong to user&quot;)
        self.run_util(
            CAGEFS_ENTER, ISOLATECTL,
            &quot;site-isolation&quot;, method, &quot;--domain&quot;, domain)
        self.exit_with_success({&quot;domain&quot;: 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(&quot;username&quot;)
            if not is_website_isolation_allowed_server_wide() or \
                    not is_website_isolation_allowed_for_user(username):
                self.exit_with_success({&quot;domainVersions&quot;: {}})
        except Exception:
            self.exit_with_success({&quot;domainVersions&quot;: {}})

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

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

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

                manager = NodeManager()
            if self.current_plugin_name == &quot;python_selector&quot;:
                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=&quot;ERROR.not_available_plugin&quot;,
                context={
                    &quot;pluginName&quot;: plugin_names.get(self.current_plugin_name, &quot;Plugin&quot;)
                },
                icon=&quot;disabled&quot;,
            )
        plugin_available_checker = {
            &quot;nodejs_selector&quot;: self._plugin_available_nodejs,
            &quot;python_selector&quot;: self._plugin_available_python,
            &quot;php_selector&quot;: self._plugin_available_php,
            &quot;resource_usage&quot;: 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=&quot;ERROR.not_available_passenger&quot;,
                context={
                    &quot;pluginName&quot;: LVEMANAGER_PLUGIN_NAMES.get(
                        self.current_plugin_name, DEFAULT_PLUGIN_NAME
                    )
                },
                icon=&quot;disabled&quot;,
            )
        if not plugin_available:
            self.exit_with_error(
                code=503,
                error_id=&quot;ERROR.not_available_plugin&quot;,
                context={
                    &quot;pluginName&quot;: LVEMANAGER_PLUGIN_NAMES.get(
                        self.current_plugin_name, DEFAULT_PLUGIN_NAME
                    )
                },
                icon=&quot;disabled&quot;,
            )

    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
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fusr%2Fshare%2Fl.v.e-manager%2Futils'>Back to Directory</a>