# 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 copy import sys import json import argparse import socket import base64 import os import subprocess import re from enum import Enum from dataclasses import dataclass from typing import Callable, List, Optional from past.builtins import basestring, unicode # noqa from future.utils import iteritems from clcommon.utils import silence_stdout_until_process_exit, get_cl_version, is_ubuntu from cllicense import CloudlinuxLicenseLib from cpanel_api import get_cpanel_api_class from clcommon.lib.cledition import is_cl_solo_edition, is_container from clcommon.cpapi import is_hitting_max_accounts_limit, get_main_username_by_uid LVEMANAGER_PLUGIN_NAMES = { "python_selector": "Python Selector", "nodejs_selector": "Node.js Selector", "php_selector": "PHP Selector", "resource_usage": "Resource Usage", "wpos": "AccelerateWP", } PASSENGER_DEPEND_PLUGINS = ["python_selector", "nodejs_selector"] DEFAULT_PLUGIN_NAME = "CloudLinux Manager" CAGEFS_ENTER_PROXIED_BIN = "/usr/bin/cagefs_enter.proxied" if not os.path.exists(CAGEFS_ENTER_PROXIED_BIN): CAGEFS_ENTER_PROXIED_BIN = "/bin/cagefs_enter.proxied" LVE_SUWRAPPER_BIN = "/usr/bin/lve_suwrapper" def parse_lve_id(value): """ Return `value` as a non-negative int when it is a usable lve id (an int or an ASCII decimal string), else None. Never raises. CLOS-6970: panel plugins send user_info['lve-id'] and it ends up as an lve_suwrapper argument; '' / ' ' / 'abc' / None / bools are not ids. ASCII only: str.isdigit() is also true for '²' or Arabic-Indic digits, which int() rejects or reads as a different number. """ if isinstance(value, bool): return None if isinstance(value, int): return value if value >= 0 else None if isinstance(value, str): text = value.strip() if text.isascii() and text.isdigit(): return int(text) return None def is_blank(value): """True for None and for strings that are empty or whitespace only.""" return value is None or (isinstance(value, str) and value.strip() == "") def is_json(data): try: json.loads(data) return True except ValueError as error: return False class EnterFlagType(Enum): """Enum for different types of enter flags""" IO_AND_MEMORY_LIMIT = "io_and_memory_limit" CPU_LIMIT = "cpu_limit" MAX_ENTER = "max_enter" class EnterTool(Enum): """Enum for different enter tools""" CAGEFS_ENTER_PROXIED = "cagefs_enter.proxied" LVE_SUWRAPPER = "lve_suwrapper" @dataclass class EnterFlags: """Abstraction layer for enter flags that can be converted between different tools""" no_io_and_memory_limit: bool = False no_cpu_limit: bool = False no_max_enter: bool = False def to_cagefs_args(self) -> List[str]: """Convert to cagefs_enter.proxied format""" args = [] if self.no_io_and_memory_limit: args.append("--no-io-and-memory-limit") if self.no_cpu_limit: args.append("--no-cpu-limit") if self.no_max_enter: args.append("--no-max-enter") return args def to_lve_suwrapper_args(self) -> List[str]: """Convert to lve_suwrapper format""" args = [] if self.no_io_and_memory_limit: args.append("-m") if self.no_cpu_limit: args.append("-c") if self.no_max_enter: args.append("-e") return args @classmethod def no_limits(cls) -> "EnterFlags": """Create flags for no limits""" return cls(no_io_and_memory_limit=True, no_cpu_limit=True, no_max_enter=True) @classmethod def all_limits(cls) -> "EnterFlags": """Create flags for all limits""" return cls(no_io_and_memory_limit=False, no_cpu_limit=False, no_max_enter=False) class CloudlinuxCliBase(object): request_data = {} result = None available_request_params = [ "owner", "command", "method", "params", "user_info", "mockJson", "attachments", "plugin_name", "lang", ] # Keys every request payload must carry. Validated by # check_required_params() from parsing_request_data(), before any # consumer dereferences them, so a malformed payload returns JSON -- the # utilities' stdout contract, which the panel plugins json_decode() -- # instead of a Python traceback (CLOS-6971). required_request_params = [ "owner", "command", ] NOT_FLAGGED_PARAMS = [ "config-files", "content", "passenger-log-file", "ignore-list", "wp-path", "upgrade-url", ] license_is_checked = False current_plugin_name = "" licence = CloudlinuxLicenseLib() def __init__(self): self.skip_cagefs_check = False self.user_info = {} self.parsing_request_data() self.check_xss() self.drop_permission() self.command_methods = { "spa-ping": self.spa_ping, "cloudlinux-top": self.cl_top, "cloudlinux-selector": self.cl_selector, "cloudlinux-statistics": self.cl_statistics, "cloudlinux-charts": self.cl_chart, "cloudlinux-quota": self.cl_quota, "cpanel-api": self.cpanel_api, "cloudlinux-xray-user-manager": self.cl_xray_user_manager, "cloudlinux-statsnotifier": self.cl_statsnotifier, "cloudlinux-awp-user": self.cloudlinux_awp_user, "cl-smart-advice-user": self.cl_smart_advice_user, "cl-install-plugin": self.cl_install_plugin, } def check_xss(self): for key in self.request_data.keys(): if key not in self.available_request_params: self.exit_with_error("BAD REQUEST 1:" + key) # Read once via .get(): "command" is a required key (see # required_request_params) but check_xss() must not raise KeyError if # it is absent -- with no command none of the per-command exemptions # below apply and the strict `else` branch is the correct fail-safe. command = self.request_data.get("command") for name, val in iteritems(self.request_data): if isinstance(val, dict): # if post key is "params" for key, inner_value in iteritems(val): self.check_param_key(key) if ( command == "cloudlinux-packages" and name == "params" and key == "package" ): self.request_data[name][key] = self.escape_param_value( inner_value ) elif command == "cloudlinux-support": pass elif ( command == "cloudlinux-selector" and name == "params" and key == "options" ): pass elif ( command == "cloudlinux-selector" and name == "params" and key == "env-vars" ): # env-vars arrives as JSON.stringify(dict). The # shell-char filter in check_param_value would # reject legitimate values like $PATH. The actual # Apache-directive-injection vector (literal \n # in value being written to .htaccess) is closed # by validate_env_vars after json.loads, plus # \r\n stripping in add_env_vars_for_htaccess. pass elif ( command == "lvectl" and name == "params" and key == "stdin" ): pass elif ( command == "cloudlinux-xray-manager" and name == "params" and key == "url" ): pass elif ( command == "cloudlinux-xray-user-manager" and name == "params" and key == "url" ): pass elif ( command == "wmt-api" and name == "params" and key == "config-change" ): pass elif ( command == "cloudlinux-xray-manager" and name == "params" and key == "email" ): pass elif ( command == "cloudlinux-awp-admin" and name == "params" and key == "upgrade-url" ): pass else: self.check_param_value(inner_value) else: self.check_param_value(val) def get_env(self): """ Get env for subprocess call """ env_copy = os.environ.copy() if self.request_data.get("lang"): lang = self.request_data.get("lang") if not re.match(r"^[a-z]{2}$", lang): lang = "en" env_copy["LC_ALL"] = lang return env_copy def get_server_ip(self): """ Get the server's IP address. """ try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) return s.getsockname()[0] except Exception as e: return None def check_param_key(self, key): # Reject any key that begins with "-": cpanel_api() renders keys # to argv with ``add_dash=False`` (libcloudlinux.py:480), so a # caller-supplied "--user" would be emitted verbatim into the # /usr/bin/uapi argv as "--user=victim", smuggling a global UAPI # option past check_operation_allowed (cpanel_api.py:231) which # only gates class/method. Requiring a leading word character # also defangs add_dash=True paths where a key like "-foo" could # be rendered as "---foo=...". if not re.search(r"^\w[\w\-]*$", key): self.exit_with_error("BAD REQUEST 2") def check_param_value(self, val): # The character set below intentionally only contains shell # metacharacters that could enable command/argument injection # when a value is passed to a subprocess. HTML/JS-relevant # characters (``<``, ``>``, ``'``, ``"``) are deliberately NOT # filtered here: this is the CLI input boundary, not a render # boundary. XSS is handled by output-escaping in the SPA and in # any cgi/template layer that emits HTML; rejecting those chars # on input would break legitimate payloads — e.g. the Options # tab posts JSON-stringified settings that contain literal # ``"``, and a previous attempt to add ``<>'"`` to this regex # (commit 1bc596507, reverted) broke reseller-tests-ui's # "Options tab → Testing saving of config" suite by making # every save return HTTP 503. if isinstance(val, basestring): if re.search("[`\|\$;&\n]", val, re.M): self.exit_with_error("BAD REQUEST 3") def escape_param_value(self, val): chars = "\\\"'" for c in chars: val = val.replace(c, "\\" + c) return val def main(self): command = self.request_data["command"] endpoint = self.command_methods.get(command) allowed_methods = [ "cloudlinux-license", "external-info", "spa-get-user-info", ] # not requires license check if endpoint: if not self.license_is_checked and command not in allowed_methods: self.check_license() if "mockJson" in self.request_data: self.spa_mock(self.request_data["mockJson"]) endpoint() else: if command: self.exit_with_error("No such module " + command) else: self.exit_with_error("Command not defined") def parsing_request_data(self): """ parsing entry data, encode it from base64 to dictionary :return: """ parser = argparse.ArgumentParser() parser.add_argument("--data") parser.add_argument("--skip-cagefs-check", action="store_true", default=False) try: arguments = parser.parse_args() except: self.exit_with_error("Unknown param in request") if arguments.data: data_in_base64 = arguments.data data_in_json = base64.b64decode(data_in_base64).decode("utf-8") try: self.request_data = json.loads(data_in_json) self.skip_cagefs_check = arguments.skip_cagefs_check except ValueError: self.exit_with_error("Need json-array") self.check_required_params() self.user_info = self.get_user_info() self.define_current_plugin() else: self.exit_with_error("No --data param in request") def check_required_params(self): """ Reject a payload that omits a required contract key. Runs before get_user_info() so the failure is reported through exit_with_error() (JSON on stdout, exit 1) rather than escaping as KeyError from whichever consumer dereferences the key first. """ missing = [ key for key in self.required_request_params if key not in self.request_data ] if missing: self.exit_with_error("BAD REQUEST 4:" + ",".join(missing)) # Scanner note (F-54, CLOS-5568, 2026-07-17): --data-supplied user_info # flagged as spoofable local-user auth-bypass. Refuted: identity here is # internal plumbing, not a privilege boundary -- every downstream target # (/var/lve/lvestats2.db 0644, /proc/lve/list 0444, /var/lve/json-stats/ # *.json 0644; /usr/sbin/cloudlinux-{statistics,top,chart} non-suid and # honor --id verbatim) is already baseline-readable to any local shell. def get_user_info(self): """ user_info as the panel plugin sent it, healed from the process identity when the plugin could not resolve it. For owner=user the worker IS the user, so the effective uid is the authoritative identity whenever the plugin's value is unusable. CLOS-6970: only None used to trigger the heal; the DirectAdmin plugin sent 'lve-id': '' (SpaManager::getUID() answered '' when ext-posix was unavailable) and str('') went into the lve_suwrapper argv as an empty lve_id. Blank and non-numeric values now heal exactly like None. The heal only makes sense when the process really is the user: under the root-side cloudlinux-cli.py (which admits owner=user for a few commands) the euid is 0, and "root" is never the right answer there - the unusable value is left alone for the callee's own validation. """ user_info = self.request_data.get("user_info") or {} euid = os.geteuid() if self.request_data.get("owner") == "user" and user_info and euid != 0 and ( any(is_blank(value) for value in user_info.values()) or ("lve-id" in user_info and parse_lve_id(user_info["lve-id"]) is None) ): username = get_main_username_by_uid(euid) user_info = {"username": username, "lve-id": euid} return user_info def cl_top(self): list_to_request = self.prepair_params_for_command() command = ["/usr/sbin/cloudlinux-top"] + list_to_request try: p = subprocess.Popen( command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) result, err = p.communicate() exitcode = p.returncode except Exception as e: self.exit_with_error( "Can't run %(command)s", context={"command": " ".join(command)}, ) return if self.request_data.get("owner") == "user": json_result = {} try: json_result = json.loads(result) except Exception: self.exit_with_error(result or err) if json_result.get("result") != "success": self.exit_with_error( json_result.get("result"), json_result.get("context"), ignore_errors=True, ) print(result) silence_stdout_until_process_exit() sys.exit(exitcode) def cl_quota(self): list_to_request = self.prepair_params_for_command() result = self.run_util( "/usr/bin/cl-quota", *list_to_request, ignore_errors=True ) print(result) def cl_xray_user_manager(self): list_to_request = self.prepair_params_for_command() list_to_request.remove("--json") result = self.run_util( "/opt/alt/php-xray/cloudlinux-xray-user-manager", *list_to_request, ignore_errors=False, ) print(result) def cl_smart_advice_user(self): cli_command = "/opt/alt/php-xray/cl-smart-advice-user" list_to_request = self.prepair_params_for_command(with_json=False) # Workaround to run the command in background if "--async" in list_to_request: subprocess.Popen( [cli_command, *list_to_request], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) self.exit_with_success() result = self.run_util(cli_command, *list_to_request, ignore_errors=False) print(result) def cpanel_api(self): owner = self.request_data.get("owner") method = self.request_data.pop("method") list_to_request = self.prepair_params_for_command( with_json=False, add_dash=False ) # For the reseller path the factory requires the caller's cPanel # username so the UAPI call is bound to that account (mirrors the # `cpapi2 --user=$CURRENT_USER` pattern in cpanel/cgi/CloudLinux.pm). # user_info is populated by the WHM CGI wrapper from REMOTE_USER / # TEAM_OWNER and is not attacker-controlled. username = (self.user_info or {}).get("username") cpanel_api = get_cpanel_api_class(owner, username=username) self.exit_with_success({"data": cpanel_api.run(method, list_to_request)}) # Keys accepted from external request for cloudlinux-charts. # Strict allowlist: lvechart's argparse permits unique-prefix abbreviations # (e.g. ``--o`` resolves to ``--output``), so a deny-by-prefix check on the # rendered argv is not sufficient. Limited to the set actually used by the # SPA chart callers; extend deliberately if a new client needs more. CHART_ALLOWED_PARAM_KEYS = frozenset(( "format", "period", "from", "to", "id", "style", "domain", )) def cl_chart(self): # cl_chart does not use ``method`` for command dispatch; if the field # is present, ``prepair_params_for_command`` would split it on spaces # and inject the resulting tokens directly into the ``lvechart`` argv # before parameter rendering, sidestepping the param-key allowlist. if "method" in self.request_data: self.exit_with_error("BAD REQUEST 2") params = self.request_data.get("params") or {} for key in params: if key not in self.CHART_ALLOWED_PARAM_KEYS: self.exit_with_error("BAD REQUEST 2") # When the caller is a panel user, pin the chart scope to the # server-authenticated lve-id (populated by the CGI wrapper from # REMOTE_USER / TEAM_OWNER, not by the client). Without this pin # a user session could smuggle ``id=`` verbatim into # the /usr/sbin/lvechart argv and read another tenant's chart # data. Mirrors the analogous pin already in cl_statistics(). if self.request_data.get("owner") == "user": params = self.request_data.setdefault("params", {}) or {} self.request_data["params"] = params params["id"] = str(self.user_info["lve-id"]) list_to_request = self.prepair_params_for_command() try: list_to_request.remove("--json") except ValueError: pass list_to_request.insert(0, "/usr/sbin/lvechart") try: response = subprocess.check_output(list_to_request, shell=False, text=True, stderr=subprocess.PIPE) except subprocess.CalledProcessError as error: self.exit_with_error((error.stderr or error.output or '').strip() or 'Chart is unavailable') print(json.dumps({"result": "success", "chart": response})) silence_stdout_until_process_exit() sys.exit(0) def drop_permission(self): """ Drop permission to users, if owner of script is user :return: """ data = self.request_data # Presence only, on purpose. A reseller is addressed by username in # the root-side CLI (--for-reseller) and is not necessarily a system # user, so panels legitimately send a non-numeric or empty lve-id for # owner=reseller (Plesk resellers: the reseller statistics page broke # with "User id does not specified" when CLOS-6970 briefly validated # the value here). The lve-id is only ever an lve_suwrapper argument # in cloudlinux_cli_user.py, which validates it itself. if data.get("owner") in ["reseller", "user"] and ( "lve-id" not in self.user_info or "username" not in self.user_info ): self.exit_with_error("User id does not specified") def prepair_params_for_command( self, with_json=True, escaped_strings=False, add_dash=True ): """ Method that converts given dict of parameters into list of strings that should be passed as arguments command-line application :param with_json: add --json argument :param escaped_strings: ONLY FOR BACKWARDS COMPATIBILITY! SHOULD BE False FOR ALL NEW METHODS! :param add_dash: if we need to add dashes to params :return: """ value_template = "--{0}={1}" if add_dash else "{0}={1}" data = copy.deepcopy(self.request_data) list_to_request = [] if "method" in data: # ``method`` is split on spaces and each token becomes its own # argv element ahead of the validated params block. A reseller # caller controlling ``method`` could otherwise smuggle e.g. # "get --user victim" past check_xss (which only filters # shell-meta values), turning the downstream selector / UAPI # argv into an attacker-controlled flag injection. Restrict # each token to subcommand-name shape: letters, digits, # dot, dash, underscore, colon (for cPanel class::method), # and explicitly no leading dash. for method in data["method"].split(" "): if not method: continue if not re.match(r"^[A-Za-z0-9_][A-Za-z0-9_.:\-]*$", method): self.exit_with_error("BAD REQUEST 2") list_to_request.append(method) if "params" not in data: data["params"] = {} if "json" not in data["params"] and with_json: data["params"]["json"] = "" for param, value in iteritems(data["params"]): if param != "additional-params": # TODO: looks like we can remove option escaped_strings # and always use value.encode('utf-8') here # same goal may be reached using utils.byteify(json.loads(...)) # but to do that, we need some tests covering unicode params # (especially for cloudlinux-packages) # unfortunately, we do not have one ;( # THIS IS NEEDED ONLY FOR CL-PACKAGES UTILITY if value and escaped_strings is True: list_to_request.append( value_template.format( param, value.encode("unicode-escape").decode() ) ) elif ( value or param in self.NOT_FLAGGED_PARAMS ) and escaped_strings is False: list_to_request.append(value_template.format(param, value)) else: list_to_request.append("--{0}".format(param)) if self.request_data["owner"] == "reseller": list_to_request.append( "--for-reseller={0}".format(self.user_info["username"]) ) if ( "additional-params" in data["params"] and data["params"]["additional-params"] != "" ): list_to_request.append("--") for param in data["params"]["additional-params"].split(): list_to_request.append("{0}".format(param)) return list_to_request def is_edition_migration_available(self): # check if edition migration is supported return os.path.isfile("/usr/sbin/clncheck") def update_license(self): # Register by broken license with open(os.devnull, "w") as devnull: clnreg_cmd = ["/usr/sbin/clnreg_ks", "--force"] if self.is_edition_migration_available(): clnreg_cmd.append("--migrate-silently") subprocess.call(clnreg_cmd, stderr=devnull, stdout=devnull, shell=False) subprocess.call( ["/usr/bin/cldetect", "--update-license"], stderr=devnull, stdout=devnull, shell=False, ) self.check_license(False) def check_license(self, with_recovery=True): if not self.kernel_is_supported(): if self.request_data["owner"] in ["reseller"]: 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", ) elif self.request_data["owner"] in ["admin"]: self.exit_with_error("Kernel is not supported") if is_hitting_max_accounts_limit(): if self.request_data["owner"] == "admin": self.exit_with_error("ERROR.hitting_max_accounts_limit") if self.request_data["owner"] == "user": 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", ) if not self.licence.get_license_status(): if self.request_data["owner"] in ["reseller", "user"]: interpreter = "nodejs" if self.request_data.get("params") and self.request_data["params"].get( "interpreter" ): interpreter = self.request_data["params"]["interpreter"] pluginNames = { "reseller": "CloudLinux Manager", "user": { "python": "Python Selector", "nodejs": "Node.js Selector", }.get(interpreter, "Node.js Selector"), } 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", ) else: if with_recovery: self.update_license() else: self.exit_with_error("License is not valid") else: self.license_is_checked = True def exit_with_error( self, error_string="", context=None, code=None, error_id=None, icon=None, ignore_errors=False, ): result = {"result": error_string} if context: result["context"] = context if code: result["code"] = code if error_id: result["error_id"] = error_id if icon: result["icon"] = icon if ignore_errors: result["ignore"] = ignore_errors print(json.dumps(result)) sys.exit(1) def exit_with_success(self, response=None): data = copy.deepcopy(response) if response else {} data["result"] = "success" print(json.dumps(data)) sys.exit(0) def cl_statistics(self): # When the caller is a panel user, pin the LVE scope to the # server-authenticated lve-id (populated by the CGI wrapper from # REMOTE_USER / TEAM_OWNER, not by the client). The reseller branch # of prepair_params_for_command already appends a server-side # `--for-reseller=` for the same utility # family — without an analogous pin for owner='user' a request body # could smuggle `id=` (or `user=`) verbatim into # the /usr/sbin/cloudlinux-statistics argv and read another tenant's # data, defeating the lve_suwrapper drop. See CHART_ALLOWED_PARAM_KEYS # for the analogous expectation that `id` is an argv-trusted scope. if self.request_data.get("owner") == "user": params = self.request_data.setdefault("params", {}) or {} self.request_data["params"] = params params["id"] = str(self.user_info["lve-id"]) params.pop("user", None) list_to_request = self.prepair_params_for_command() command = ["/usr/sbin/cloudlinux-statistics"] + list_to_request try: p = subprocess.Popen( command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) result, err = p.communicate() except Exception as e: self.exit_with_error( "Can't run %(command)s", context={"command": " ".join(command)}, ) return result = self.adjust_statistics_output(result) if result: print(result, end="") if p.returncode != 0 and err: print(err, end="", file=sys.stderr) silence_stdout_until_process_exit() sys.exit(p.returncode) def adjust_statistics_output(self, output): """Hook for panel-specific rewriting of cloudlinux-statistics output.""" return output def spa_mock(self, file): file_path = "/usr/share/l.v.e-manager/spa/src/jsons/%s.json" % (file) # check if passed file param doesn't use relative path. E.g.: '../../file' if os.path.realpath(file_path) != file_path: self.exit_with_error("BAD REQUEST 3") with open(file_path, "r") as f: print(f.read()) sys.exit(0) def get_lve_version(self): try: ver = subprocess.check_output( "cat /proc/lve/list | grep -Po '^\d{1,2}:'", shell=True, executable="/bin/bash", text=True, ).strip() return int(ver[:-1]) except: return 0 def get_cloudlinux_version(self): return subprocess.check_output( "uname -r | grep -Po 'el\d\w?'", shell=True, executable="/bin/bash", text=True, ).strip() # Common methods def spa_ping(self): self.exit_with_success() def cl_selector(self): try: from clselector.cl_selector import CloudlinuxSelector except Exception as e: self.exit_with_error(f"Module unavailable: {e}") if ( self.user_info.get("username") and "interpreter" in self.request_data["params"] and self.request_data["params"]["interpreter"] == "php" ): self.check_php_selector_user_availablility() list_to_request = self.prepair_params_for_command() cll = CloudlinuxSelector() # ``cloudlinux-selector`` returns a status rather than exiting for # ordinary command failures. This process is the cPanel endpoint # after the CageFS re-exec, so dropping a nonzero value turns a failed # npm resolution into an HTTP-successful command. Keep successful # commands on their established fall-through path. exit_code = cll.run(list_to_request) if exit_code: sys.exit(exit_code) def check_php_selector_user_availablility(self): """ Additional check only for php selector :return: """ try: LIBDIR = "/usr/share/cagefs" sys.path.append(LIBDIR) import cagefsctl if not cagefsctl.cagefs_is_enabled or not cagefsctl.is_user_enabled( self.user_info["username"] ): raise RuntimeError("Cagefs is disabled or missing") except (ImportError, RuntimeError): self.exit_with_error( code=503, error_id="ERROR.cagefsDisabled", ) from clselect.clselectexcept import BaseClSelectException try: from clselect import ClSelect ClSelect.check_multiphp_system_default_version() except BaseClSelectException: self.exit_with_error( code=503, error_id="ERROR.systemVersionAltPHP", ) def define_current_plugin(self): self.current_plugin_name = self.request_data.get("plugin_name") def is_error_response_default(self, json_result): return ( json_result.get("result") != "success" and json_result.get("success") != 1 ) def run_util(self, name, *args, **kwargs): command = [name] + list(args) error_checker = kwargs.get("error_checker", self.is_error_response_default) try: p = subprocess.Popen( command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=kwargs.get("env", self.get_env()), ) (result, err) = p.communicate(kwargs.pop("stdin", None)) is_error = p.returncode != 0 or not is_json(result) if not is_error: json_result = json.loads(result) is_error = error_checker(json_result) if is_error: result = result + err if is_json(result): json_result = json.loads(result) if json_result.get("message"): json_result["result"] = json_result.pop("message") result = json.dumps(json_result) if kwargs.get("ignore_errors", False): # Check new result concatenated with error if is_json(result): result = json.loads(result) result["ignore"] = True result = json.dumps(result) else: result = self.ignored_error_message(result) print(result) exit(1) return result except Exception as e: self.exit_with_error( "Can't run %(command)s", context={"command": " ".join(command)}, ignore_errors=True, ) def ignored_error_message(self, message): return json.dumps({"result": message, "ignore": True}) def kernel_is_supported(self): try: if is_container(): return True if is_cl_solo_edition(skip_jwt_check=True): # Ubuntu uses standard kernel, # CL9 uses Alma kernel which doesn't have 'lve' in its name if is_ubuntu() or get_cl_version() in ["cl9", "cl10"]: return True uname = subprocess.check_output( "uname -r", shell=True, executable="/bin/bash", text=True ) return "lve" in uname else: f = open("/proc/lve/list", "r") line = f.readline() f.close() return bool(line) except IOError: return False def cl_statsnotifier(self): list_to_request = self.prepair_params_for_command(with_json=True) command = ["/usr/sbin/cloudlinux-statsnotifier"] + list_to_request try: p = subprocess.Popen( command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) result, err = p.communicate() except Exception as e: self.exit_with_error( "Can't run %(command)s", context={"command": " ".join(command)}, ) return if result: print(result, end="") if p.returncode != 0 and err: print(err, end="", file=sys.stderr) sys.exit(p.returncode) def cloudlinux_awp_user(self): cli_command = "/usr/bin/cloudlinux-awp-user" list_to_request = self.prepair_params_for_command(with_json=False) result = self.run_util(cli_command, *list_to_request, ignore_errors=False) print(result) def cl_install_plugin(self): """ This method is needed just for dev server to allow work with mocks """ self.exit_with_success() # user cli parts class CommandType(Enum): HEAVY = "heavy" SIMPLE = "simple" class ConfigLimitValue(Enum): ALL = "all" # limit all requests HEAVY = "heavy" # don't limit white-listed 'simple' requests UNLIMITED = "unlimited" # don't limit at all @classmethod def _missing_(cls, value): return cls.ALL @dataclass class Rule: callable: Callable result: CommandType class LimitStrategyBase: """ Base limits strategy to decide - run incoming request with or without cagefs limits Strategy execution means that used script (cloudlinux_cli_user.py) will be re-executed with (or not) additional enter flags for the selected utility """ enter_flags: EnterFlags enter_tool: EnterTool = ( EnterTool.CAGEFS_ENTER_PROXIED ) # Default to cagefs_enter.proxied def execute( self, uid: int, command: str, args: List[str], request_data: dict ) -> Optional[int]: full_command = self.get_full_command(uid, command, args, request_data) p = subprocess.Popen(full_command) p.communicate() return p.returncode def get_full_command( self, uid: int, command: str, args: List[str], request_data: dict ) -> List[str]: cmd = [*sys.argv, f"--skip-cagefs-check"] if self.enter_tool == EnterTool.LVE_SUWRAPPER: lve_id = parse_lve_id(uid) if lve_id is None: # CLOS-6970: str('') is '' — lve_suwrapper would receive an # empty lve_id and print its usage text instead of running # the command. Callers validate earlier; this is the last line. raise ValueError( "refusing to run lve_suwrapper with an unusable lve_id: %r" % (uid,) ) return [ LVE_SUWRAPPER_BIN, "-n", # bypass cagefs namespace *self.enter_flags.to_lve_suwrapper_args(), str(lve_id), *cmd, ] else: return [CAGEFS_ENTER_PROXIED_BIN, *self.enter_flags.to_cagefs_args(), *cmd] class BypassStrategy(LimitStrategyBase): """ Strategy which bypasses both lve and cagefs and executes commands direcly. """ def execute( self, uid: int, command: str, args: List[str], request_data: dict ) -> Optional[int]: return class AllLimitStrategy(LimitStrategyBase): """ Strategy to limit all commands """ enter_flags = EnterFlags.all_limits() class NoLimitStrategy(LimitStrategyBase): """ Strategy to don't limit all commands """ enter_flags = EnterFlags.no_limits() class LimitStrategyHeavy(LimitStrategyBase): """ Strategy to don't limit whitelisted commands By default - all commands are HEAVY and will be limited Add `rules` to mark any command as SIMPLE and run without limits """ enter_flags = EnterFlags.all_limits() default_rule = Rule(callable=lambda args: True, result=CommandType.HEAVY) rules = { "cloudlinux-selector": [ Rule(callable=lambda args: "get" in args, result=CommandType.SIMPLE), Rule(callable=lambda args: "start" in args, result=CommandType.SIMPLE), Rule(callable=lambda args: "restart" in args, result=CommandType.SIMPLE), Rule(callable=lambda args: "stop" in args, result=CommandType.SIMPLE), ], # the following lve-stats utilities allow user to monitor his load # they should be available disregarding the load on the server "cloudlinux-top": [ Rule(callable=lambda args: True, result=CommandType.SIMPLE), ], "cloudlinux-statistics": [ Rule(callable=lambda args: True, result=CommandType.SIMPLE), ], "cloudlinux-charts": [ Rule(callable=lambda args: True, result=CommandType.SIMPLE), ], "spa-get-user-info": [ Rule(callable=lambda args: True, result=CommandType.SIMPLE), ], } def _check_rules(self, command: str, args: List[str]) -> CommandType: command_type = None for rule in self.rules.get(command, []) + [self.default_rule]: if rule.callable(args): command_type = rule.result break if command_type == CommandType.SIMPLE: self.enter_flags = EnterFlags.no_limits() else: self.enter_flags = EnterFlags.all_limits() return command_type def get_full_command( self, uid: int, command: str, args: List[str], request_data: dict ) -> List[str]: self._check_rules(command, args) return super().get_full_command(uid, command, args, request_data)