File manager - Edit - /home/sarjilsh/cityfoodcenter.co.uk/l.v.e-manager.tar
Back
utils/set_env_vars.py 0000644 0000000 0000000 00000010016 00000000000 012153 0 ustar 00 #!/opt/cloudlinux/venv/bin/python3 -bb # -*- coding: utf-8 -*- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from __future__ import print_function from __future__ import division from __future__ import absolute_import import os import sys import getpass from future.utils import iteritems from clselect.clselectnodejs.apps_manager import ApplicationsManager as NodeJsAppsManager from clselect.clselectpython.apps_manager import ApplicationsManager as PythonAppsManager, get_venv_rel_path from clselect.utils import get_using_realpath_keys def get_app_name(interpreter): """ Get application name via CL_APP_ROOT variable :param interpreter: interpreter name :return: str application name """ if os.environ.get('CL_APP_ROOT'): return os.environ['CL_APP_ROOT'] elif interpreter == 'python': abs_venv_path = os.environ['VIRTUAL_ENV'] user_home = os.environ['HOME'] # /home/<username>/virtualenv/<name>/<version> -> # /virtualenv/<name> vevn_rel_path = os.path.dirname(abs_venv_path).replace(user_home + '/', '', 1) # if VENV_REL_PATH contains _ we cannot # clearly define app_root and should guess if '_' in vevn_rel_path: # scanner-triage: set_env_vars.py hard-rejects root at __main__ via is_root() # and is installed 0755 with no setuid, so a spoofed getpass.getuser() only # reads files with the attacker's own UID — no elevation. User-side interpreter # invocations also run inside CageFS where other users' /home is invisible. username = getpass.getuser() # in python CL_APP_ROOT is not set, we must guess by env path for app_root in PythonAppsManager().get_user_config_data(username): _, rel_path = get_venv_rel_path(username, app_root) if rel_path == vevn_rel_path: return app_root return None else: return vevn_rel_path.replace('virtualenv/', '', 1) else: raise NotImplementedError( 'I don\'t know how to get app_root for %s' % interpreter) def get_env_vars(_app_name, interpreter): """ Get environment variables from user config for given application name :param _app_name: application name :param interpreter: interpreter name :return: dict {ENV_VAR_NAME: VALUE} """ _env_vars = {} username = getpass.getuser() try: full_app_config = get_app_full_conf(username, _app_name, interpreter) if interpreter == 'nodejs': _env_vars['NODE_ENV'] = full_app_config['app_mode'] _env_vars.update(full_app_config['env_vars']) except KeyError: pass return _env_vars def set_env_vars(dict_env_vars): """ Print to stdout bash strings with environment variables :param dict_env_vars: dict with environment variables :return: None """ for key, var in iteritems(dict_env_vars): print('export {}="{}"'.format(key, var)) def is_root(): return os.geteuid() == 0 def get_app_full_conf(user, app, interpreter): if interpreter == 'nodejs': manager = NodeJsAppsManager() elif interpreter == 'python': manager = PythonAppsManager() else: raise NotImplementedError() full_user_config = manager.get_user_config_data(user) if not full_user_config: print("User config was not found or empty") sys.exit(0) return get_using_realpath_keys(user, app, full_user_config) if __name__ == "__main__": if is_root(): print("This program is not intended to be run as root.") sys.exit(1) args = sys.argv if len(args) < 2: print("Interpreter is not passed.") sys.exit(1) app_name = get_app_name(sys.argv[1]) if app_name is None: print("Unknown application.") sys.exit(1) env_vars = get_env_vars(app_name, sys.argv[1]) set_env_vars(env_vars) utils/cloudlinux_cli_user.py 0000644 0000000 0000000 00000050006 00000000000 013533 0 ustar 00 # 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" ISOLATECTL = "/usr/sbin/isolatectl" 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 Exception as e: self.exit_with_error(f"Module unavailable: {e}") sys.exit(0) def get_user_domains(self): try: from clcommon.cpapi import userdomains except Exception as e: self.exit_with_error(f"Module unavailable: {e}") 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 # 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 "denied by your server # administrator"). 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 "status unavailable" 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, "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": # 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("result") or (p.stderr or p.stdout or "").strip() or "site-isolation list exited with code %s" % p.returncode ) else: isolated = set(list_result.get("enabled_sites", [])) 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 = [ { "domain": d, "isolated": d in isolated, "selectorCompatible": d in selector_domains, } for d in self.get_user_domains() ] result = { "featureAvailable": feature_available, "statusAvailable": status_available, "allowed": user_allowed, "domains": domains_info, } if status_error: result["statusError"] = status_error self.exit_with_success(result) def _site_isolation_toggle(self, method, params): domain = params.get("domain") if not domain: self.exit_with_error("Missing domain parameter") # 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("username") if not is_website_isolation_allowed_server_wide() or \ not is_website_isolation_allowed_for_user(username): self.exit_with_error("Site isolation is not allowed for this user") 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("Site isolation feature is unavailable") 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, ISOLATECTL, "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, ISOLATECTL, "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 utils/cloudlinux-cli-user.py 0000644 0000000 0000000 00000000753 00000000000 013373 0 ustar 00 #!/opt/cloudlinux/venv/bin/python3 -sbb # 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 from cloudlinux_cli_user import CloudlinuxCliUser if __name__ == "__main__": cloudlinux_cli = CloudlinuxCliUser() cloudlinux_cli.main() utils/python_wrapper 0000644 0000000 0000000 00000001070 00000000000 012107 0 ustar 00 #!/bin/bash # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # if [[ $EUID -eq 0 ]]; then echo "This program is not intended to be run as root." 1>&2 exit 1 fi CWD=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) source ${CWD}/activate CL_PYTHON_VERSION="$(echo "${VIRTUAL_ENV}" | awk -F '/' '{print $NF}')" eval $(${CWD}/set_env_vars.py python) ABSOLUTE_PATH="${CWD}/python${CL_PYTHON_VERSION}_bin" exec "${ABSOLUTE_PATH}" "$@" utils/libcloudlinux.py 0000644 0000000 0000000 00000000000 00000000000 012322 0 ustar 00 utils/npm_wrapper 0000644 0000000 0000000 00000003302 00000000000 011360 0 ustar 00 #!/bin/bash if [[ $EUID -eq 0 ]]; then echo "This program is not intended to be run as root." 1>&2 exit 1 fi error_msg="Cloudlinux NodeJS Selector demands to store node modules for application in separate folder \ (virtual environment) pointed by symlink called \"node_modules\". That's why application should not contain \ folder/file with such name in application root" CWD=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) source "${CWD}/activate" eval $(${CWD}/set_env_vars.py nodejs) app_node_modules="${HOME}/${CL_APP_ROOT}/node_modules" venv_node_modules="${CL_VIRTUAL_ENV}/lib/node_modules" nodejs_npm="$CL_NODEHOME/usr/bin/npm" # install with its aliases and list with its alias without arguments or +args if [[ "$@" =~ ^(install|i|add|list|la|ll)$ || "$@" =~ ^(install|i|add|list|la|ll)[[:space:]].*$ ]]; then # We remove old symlink `~/app_root/node_modules` if it exists if [[ -L "${app_node_modules}" ]]; then rm -f "${app_node_modules}" || (echo "Can't remove symlink "${app_node_modules} 1>&2 && exit 1) # We print error end exit 1 if `~/app_root/node_modules` is dir or file elif [[ -d "${app_node_modules}" || -f "${app_node_modules}" ]]; then echo "${error_msg}" 1>&2 && exit 1 fi # we should create venv/node_modules, https://docs.cloudlinux.com/index.html?link_traversal_protection.html mkdir -p "${venv_node_modules}" # Create symlink ~/app_root/node_modules to ~/nodevenv/app_root/int_version/lib/node_modules ln -fs "${venv_node_modules}" "${app_node_modules}" ln -sf "${HOME}/${CL_APP_ROOT}/package.json" "${CL_VIRTUAL_ENV}/lib/package.json" exec "${nodejs_npm}" "$@" --prefix="${CL_VIRTUAL_ENV}/lib" else exec "${nodejs_npm}" "$@" fi utils/cpanel_api.py 0000644 0000000 0000000 00000026614 00000000000 011563 0 ustar 00 # coding:utf-8 # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from __future__ import absolute_import from __future__ import print_function from lvemanager.helpers import exit_with_error, run_command import json import re from abc import abstractmethod from clcommon.lib.whmapi_lib import WhmApiRequest, WhmApiError import sys OWNER_ADMIN = 'admin' OWNER_RESELLER = 'reseller' OWNER_USER = 'user' UAPI_IGNORED_ERRORS = ["The event UAPI::LangPHP::php_set_vhost_versions was handled successfully."] def get_cpanel_api_class(owner, username=None): """ Factory method that returns certain class depending on owner's type :param owner: Can be 'admin', 'reseller' or 'user' :param username: required when owner == 'reseller' so the UAPI call is bound to the reseller's cPanel account :return: """ if owner == OWNER_ADMIN: return CPanelAdminApi() if owner == OWNER_RESELLER: # Reseller path runs as root (see drop_permission in # cloudlinux_cli.py:157); UAPI inherits no implicit scope from # cpsrvd env, so the username must be passed explicitly so the # call is bound to the reseller's own cPanel account. if not username: exit_with_error('Reseller cpanel-api call requires username') return CPanelUserApi(scoped_user=username) return CPanelUserApi() class CPanelApi(object): """ Abstract method that defines abstract methods that must be implemented and common methods of the derived classes """ RETURN_JSON = '--output=json' ALLOWED_OPERATIONS = {} @abstractmethod def check_operation_allowed(self, called_method): """ Checks whether operation is allowed to perform :param called_method: :return: """ raise NotImplementedError('Method check_operation_allowed must be implemented!') @abstractmethod def prepare_running_command(self, called_method, params, return_json): """ Depending on the passed arguments, the method builds running command """ raise NotImplementedError('Method prepare_running_command must be implemented!') @abstractmethod def parse_result(self, called_method, result): """ Some results are parsed before sending back to the called. This method parses it and returns transformed result. """ raise NotImplementedError('Method parse_result must be implemented!') @staticmethod def check_method_allowed(method_name, methods): """ CPanel API contains several methods and not all of them if supported by this utility. """ for method_object in methods: if method_name == method_object['method']: return True return False def check_for_errors(self, result): """ Checks whether CPanel API returned error or not :param result: :return: """ raise NotImplementedError('Method check_for_errors must be implemented!') @staticmethod def get_method_parser(method_name, methods): """ Some result must be parsed before they are sent back to the caller. This method extracts the method that must be executed upon the result to transform it. """ for method_object in methods: if method_name == method_object['method']: return method_object.get('parser', None) def get_ignore_errors(self, method_name, methods): """ return ignore error flag for method :return: """ for method_object in methods: if method_name == method_object['method']: return method_object.get('ignore_errors', None) @staticmethod def parse_vhost_versions(result): """ Method parses the result of the method 'parse_vhost_versions'. The host will be inherited when its field 'sys_default' inside 'php_version_source' is equal to 1. """ parsed_result = [] for r in result: parsed_result.append({ 'version': r.get('version'), 'host': r.get('vhost'), 'php_fpm': True if r.get('php_fpm') else False, 'inherited': True if r.get('php_version_source', {}).get('sys_default') == 1 else False }) return parsed_result def run(self, called_method, params, return_json=True): """ Default method which first checks whether operation is allowed then if allowed runs prepared command. After receiving the result, it will call parse_result method to transform result into desired format. """ if self.check_operation_allowed(called_method): prepared_command = self.prepare_running_command(called_method, params, return_json) code, output, std_err = run_command(prepared_command, return_full_output=True) if code != 0: exit_with_error(std_err or 'output of the command: %s\n%s' % (prepared_command, output)) try: result = json.loads(output) return self.parse_result(called_method, result) except ValueError as e: exit_with_error(e) else: exit_with_error('Not allowed operation: {}'.format(called_method)) class CPanelAdminApi(CPanelApi): COMMAND = '/usr/local/cpanel/bin/whmapi1' ALLOWED_OPERATIONS = [ { 'method': 'php_get_vhost_versions', 'parser': lambda result: CPanelAdminApi.parse_vhost_versions(result) }, {'method': 'php_get_system_default_version'}, {'method': 'php_set_vhost_versions'}, {'method': 'php_get_installed_versions'} ] def check_operation_allowed(self, called_method): return self.check_method_allowed(called_method, self.ALLOWED_OPERATIONS) def prepare_running_command(self, called_method, params, return_json): transformed_params = {} for param in params: key, value = param.split('=') transformed_params[key] = value return transformed_params def parse_result(self, called_method, result): parser = self.get_method_parser(called_method, self.ALLOWED_OPERATIONS) if parser is not None: return parser(result) return result @staticmethod def parse_vhost_versions(result): parsed_result = [] for r in result['versions']: parsed_result.append({ 'version': r.get('version'), 'host': r.get('vhost'), 'user': r.get('account'), 'php_fpm': True if r.get('php_fpm') else False, 'inherited': True if r.get('php_version_source', {}).get('sys_default') == 1 else False }) return parsed_result def run(self, called_method, params, return_json=True): params = self.prepare_running_command(called_method, params, return_json) if self.check_operation_allowed(called_method): try: return self.parse_result(called_method, WhmApiRequest(called_method).with_arguments(**params).call()) except WhmApiError as e: print(e) sys.exit(1) else: exit_with_error('Not allowed operation: {}'.format(called_method)) class CPanelUserApi(CPanelApi): COMMAND = '/usr/bin/uapi' USER_BINDING_RE = re.compile(r'^[a-zA-Z][a-zA-Z0-9_-]*$') ALLOWED_OPERATIONS = { 'LangPHP': [ {'method': 'php_set_vhost_versions'}, { 'method': 'php_get_installed_versions', 'ignore_errors': True, }, { 'method': 'php_get_vhost_versions', 'parser': lambda result: CPanelApi.parse_vhost_versions(result), 'ignore_errors': True, }, { 'method': 'php_get_system_default_version', 'ignore_errors': True, } ] } def __init__(self, scoped_user=None): # scoped_user pins the UAPI call to a specific cPanel account by # prepending `--user=<scoped_user>` to argv. Set for the reseller # path (see get_cpanel_api_class) so a reseller cannot influence # state on accounts other than their own. Left None for the # `owner == 'user'` path where the process already drops uid to # the user and UAPI infers scope from kernel identity. if scoped_user is not None and not self.USER_BINDING_RE.match(scoped_user): exit_with_error('Invalid scoped_user: {}'.format(scoped_user)) self.scoped_user = scoped_user def check_class_allowed(self, class_name): """ CPanel UAPI requires to pass class name and only few of classes are supported by this utility. """ return class_name in self.ALLOWED_OPERATIONS.keys() def check_operation_allowed(self, called_method): """ Checks whether operation is allowed or not. By CPanel UAPI supports several operations but only few of them can be called via this utility. """ class_name, method_name = self.extract_class_and_method(called_method) if self.check_class_allowed(class_name): return self.check_method_allowed(method_name, self.ALLOWED_OPERATIONS[class_name]) return False @staticmethod def extract_class_and_method(called_method): try: class_name, method_name = called_method.split('::') return class_name, method_name except ValueError: exit_with_error('Invalid method is passed: {}'.format(called_method)) def prepare_running_command(self, called_method, params, return_json): class_name, method_name = self.extract_class_and_method(called_method) argv = [self.COMMAND] if self.scoped_user: # Bind the call to a specific cPanel account; mirrors the # `cpapi2 --user=$CURRENT_USER` shape used by # cpanel/cgi/CloudLinux.pm:282-283 for the same reason. argv.append('--user=' + self.scoped_user) argv += [class_name, method_name] + params if return_json: argv.append(self.RETURN_JSON) return argv def parse_result(self, called_method, result): class_name, method_name = self.extract_class_and_method(called_method) ignore_errors = self.get_ignore_errors(method_name, self.ALLOWED_OPERATIONS[class_name]) self.check_for_errors(result, ignore_errors) parser = self.get_method_parser(method_name, self.ALLOWED_OPERATIONS[class_name]) if parser is not None: return parser(result['result']['data']) return result['result']['data'] def check_for_errors(self, result, ignore_errors = False): errors = result['result'].get('errors') if not errors: return for error in errors: # TODO: The next line is workaround of cPanel issue CPANEL-35122 # see https://support.cpanel.net/hc/en-us/articles/1500004317181-PHP-Selector-change-to-CloudLinux-PHP-version-reports-a-false-error if error in UAPI_IGNORED_ERRORS: continue exit_with_error(' '.join(errors), ignore_errors=ignore_errors) utils/node_wrapper 0000644 0000000 0000000 00000000414 00000000000 011514 0 ustar 00 #!/bin/bash if [[ $EUID -eq 0 ]]; then echo "This program is not intended to be run as root." 1>&2 exit 1 fi CWD=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) source ${CWD}/activate eval $(${CWD}/set_env_vars.py nodejs) exec "${CL_NODEHOME}/usr/bin/node" "$@" utils/cloudlinux-selector.py 0000644 0000000 0000000 00000001216 00000000000 013463 0 ustar 00 #!/opt/cloudlinux/venv/bin/python3 -sbb # -*- coding: utf-8 -*- # cloudlinux-selector Utility to check Cloudlinux license # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 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 sys from clselector.cl_selector import CloudlinuxSelector def main(argv): """ Main run function """ cll = CloudlinuxSelector() return cll.run(argv) if __name__ == "__main__": sys.exit(main(sys.argv[1:])) utils/activate 0000644 0000000 0000000 00000004461 00000000000 010635 0 ustar 00 #!/bin/bash if [[ x"${BASH_SOURCE[0]}" == x"$0" ]]; then echo "'activate' script should be sourced, not run directly" exit 1; fi CWD=$(cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd) NEW_VIRTUAL_ENV_PATH="${CWD%/bin}" deactivate () { if [[ x"${BASH_SOURCE[0]}" == x"$0" ]]; then echo "'deactivate' script should be sourced, not run directly" exit 1; fi # Only restore from backup variables if they are set # But include the case when they are set to be empty if [[ ${BKP_PATH+"is_set"} == "is_set" ]]; then PATH="$BKP_PATH" export PATH fi if [[ ${BKP_NODE_PATH+"is_set"} == "is_set" ]]; then NODE_PATH="$BKP_NODE_PATH" export NODE_PATH fi if [[ ! -z $BKP_PS1 ]]; then PS1="$BKP_PS1" export PS1 fi unset -v BKP_PATH unset -v BKP_NODE_PATH unset -v BKP_PS1 unset -v CL_VIRTUAL_ENV unset -v CL_APP_ROOT unset -v CL_NODEHOME unset -v CL_NODEJS_VERSION if [ ! "${1-}" = "nondestructive" ] ; then # Self destruct! unset -f deactivate fi } activate () { CL_VIRTUAL_ENV="${NEW_VIRTUAL_ENV_PATH}" CL_NODEJS_VERSION="$(echo "${CL_VIRTUAL_ENV}" | awk -F '/' '{print $NF}')" CL_APP_ROOT="${CL_VIRTUAL_ENV#$HOME/nodevenv/}" # cut $HOME/nodevenv/ CL_APP_ROOT="${CL_APP_ROOT%/$CL_NODEJS_VERSION}" # cut nodejs version CL_NODEHOME="/opt/alt/alt-nodejs${CL_NODEJS_VERSION}/root" BKP_NODE_PATH="$NODE_PATH" NODE_PATH="$CL_VIRTUAL_ENV/lib/node_modules:$CL_NODEHOME/lib/node_modules:$CL_NODEHOME/lib64/node_modules:$NODE_PATH" BKP_PATH="$PATH" PATH="$CL_VIRTUAL_ENV/bin:$CL_NODEHOME/usr/bin:$CL_VIRTUAL_ENV/lib/bin/:$PATH" if [[ -z "$CL_VIRTUAL_ENV_DISABLE_PROMPT" ]] ; then BKP_PS1="$PS1" PS1="[${CL_APP_ROOT} ($CL_NODEJS_VERSION)] $PS1" fi export BKP_PS1 export BKP_PATH export BKP_NODE_PATH export PS1 export CL_VIRTUAL_ENV export CL_APP_ROOT export CL_NODEHOME export NODE_PATH export PATH } # compare current virtual environment (that is stored in CL_VIRTUAL_ENV) path # to the NEW_VIRTUAL_ENV_PATH, that is the path of the new environment we may enter too # just do nothing if paths are equal if [ "${CL_VIRTUAL_ENV}" != "${NEW_VIRTUAL_ENV_PATH}" ]; then deactivate nondestructive activate fi cl.nodejs/yum_cache.dat 0000644 0000000 0000000 00000000246 00000000000 012255 0 ustar 00 alt-nodejs10 alt-nodejs11 alt-nodejs12 alt-nodejs14 alt-nodejs16 alt-nodejs18 alt-nodejs19 alt-nodejs20 alt-nodejs22 alt-nodejs24 alt-nodejs6 alt-nodejs8 alt-nodejs9 cl.python/yum_cache.dat 0000644 0000000 0000000 00000000240 00000000000 012306 0 ustar 00 alt-python27 alt-python33 alt-python34 alt-python35 alt-python36 alt-python37 alt-python38 alt-python39 alt-python310 alt-python311 alt-python312 alt-python313 cl.python/selector.json 0000644 0000000 0000000 00000000041 00000000000 012371 0 ustar 00 { "selector_enabled": false } lvemanager-config.json 0000644 0000000 0000000 00000000450 00000000000 012223 0 ustar 00 {"ui_config": {"inodeLimits": {"showUserInodesUsage": true}, "uiSettings": {"hideRubyApp": true, "hideLVEUserStat": false, "hidePythonApp": true, "hideNodeJsApp": false, "hidePHPextensions": false, "hideDomainsTab": false, "hidePhpApp": false, "hideXrayApp": false, "hideAccelerateWPApp": true}}}
| ver. 1.4 |
Github
|
.
| PHP 8.4.23 | Generation time: 0 |
proxy
|
phpinfo
|
Settings