??????????????????????????????? ??????????????????????????????? ??????????????????????????????? >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<< <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<< >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>. ??????????????????????????????? ??????????????????????????????? ??????????????????????????????? ??????????????????????????????? ??????????????????????????????? ??????????????????????????????? >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<< <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<< >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>. >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<< PK!~ß¡` cl.python/selector.jsonnu„[µü¤{ "selector_enabled": true }PK!‰·y  cl.python/yum_cache.datnu„[µü¤alt-python27 alt-python33 alt-python34 alt-python35 alt-python36 alt-python37 alt-python38 alt-python39 alt-python310 alt-python311 alt-python312 alt-python313 PK!}¾™ ±(±("panelless-version/daemon/server.pynuȯÝí#!/opt/cloudlinux/venv/bin/python3 -bb # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import base64 import json import logging import os import uuid from encodings.base64_codec import base64_encode from pwd import getpwnam import aiohttp_jinja2 import jinja2 import pam from aiohttp import web from aiohttp.web_request import Request from aiohttp.web_response import Response from aiohttp_session import setup as setup_session from aiohttp_security import is_anonymous, remember, forget, \ setup as setup_security, SessionIdentityPolicy, authorized_userid, check_permission from aiohttp_security.abc import AbstractAuthorizationPolicy from aiohttp_session.cookie_storage import EncryptedCookieStorage from cryptography import fernet from constants import * from utils import parse_params, run_cmd_pw, get_user_data, get_service_port,\ get_ssl_context, get_user_type, get_user_plugins # ToDo(dpoleev): Use PKG_VERSION from lvemanager package after migrate to python 3.7 try: PKG_VERSION = open('/usr/share/l.v.e-manager/version').read().strip() except (FileNotFoundError, OSError): # File missing in dev/test contexts where the package isn't installed. PKG_VERSION = '0.0.0-dev' logger = logging.getLogger(__name__) SERVER_PATH = os.path.dirname(os.path.realpath(__file__)) STATIC_FILE_PATH = os.path.join(SERVER_PATH, '../../commons/spa-resources/') CLOUDLINUX_CLI = '/usr/share/l.v.e-manager/utils/cloudlinux-cli.py' CLOUDLINUX_USER_CLI = '/usr/share/l.v.e-manager/utils/cloudlinux-cli-user.py' DISABLE_CSP_FLAG='/var/lve/disable_csp.flag' SECURITY_HEADERS = { "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY", "Content-Security-Policy": "default-src 'self' 'unsafe-inline' *.cloudlinux.com *.googleapis.com;object-src 'none';font-src *;script-src 'self' 'unsafe-eval' 'unsafe-inline' *.cloudlinux.com;img-src 'self' data:" } class AuthorizationPolicy(AbstractAuthorizationPolicy): """ get information about user by identity """ async def authorized_userid(self, identity): pw = getpwnam(identity) result = { 'pw': pw, 'type': get_user_type(pw.pw_name) } return result async def permits(self, identity, permission, context=None): if not identity: return False if permission not in PLUGINS: return False if permission == 'main' and get_user_type(identity) not in (TYPE_ADMIN, TYPE_RESELLER): return False return True async def handler_root(request): """ Main page """ is_logged = not await is_anonymous(request) if is_logged: user_data = await authorized_userid(request) if user_data['type'] == TYPE_ADMIN: return await admin_page(request) elif user_data['type'] == TYPE_RESELLER: return await reseller_page(request) else: return await user_page(request) else: return await login_page(request) @aiohttp_jinja2.template('login.html') async def login_page(request: Request): """ Login page """ return { 'login_error': request.query_string == 'incorrect' } @aiohttp_jinja2.template('user.html') async def user_page(request): """ List applications page """ return { 'userdata': await authorized_userid(request), 'apps': get_user_plugins() } async def admin_page(request): """ Open admin LveManager """ response = await show_app('main', await authorized_userid(request), request) set_csrf_token(response) return response async def reseller_page(request): """ Open Reseller LveManager """ response = await show_app('main', await authorized_userid(request), request) set_csrf_token(response) return response async def handler_login(request: Request): """ Check authorization and show error message """ redirect_response = web.HTTPFound('/') data = await request.post() pam_object = pam.pam() if pam_object.authenticate(data.get('username'), data.get('password'), service='system-auth'): await remember(request, redirect_response, data.get('username')) return redirect_response else: return web.HTTPFound('/?incorrect') async def handler_app(request: Request): """ Show app handler """ plugin_name = request.match_info['plugin_name'] await check_permission(request, plugin_name) userdata = await authorized_userid(request) response = await show_app(plugin_name, userdata, request) set_csrf_token(response) return response @aiohttp_jinja2.template('app.html') async def show_app(plugin_name, userdata, request: Request): """ Show certain SPA application :param plugin_name: plugin name show what bundle and title should be used :param userdata: information about user :param request:Request for rendering jinja template :return: """ plugin_title = PLUGINS.get(plugin_name).get('title') panel_data = await get_user_data(userdata) return { 'plugin_title': plugin_title, 'plugin_name': plugin_name, 'panel_data': panel_data, 'pluginVersion': PKG_VERSION } async def handler_logout(request): """ Logout handler: remove cookies information """ redirect_response = web.HTTPFound('/') await forget(request, redirect_response) return redirect_response async def handler_request(request): """ Middleware for providing requests to cloudlinux-cli level """ check_csrf_token(request) plugin_name = request.match_info['plugin_name'] plugin_title = PLUGINS.get(plugin_name).get('title') await check_permission(request, plugin_name) user_data = await authorized_userid(request) try: request_data = parse_params(await request.post()) except ValueError: raise web.HTTPBadRequest(body='Bad request') data = { 'plugin_name': plugin_name, 'owner': user_data['type'], 'command': request_data.get('command'), 'params': request_data.get('params') or {}, } if 'mockJson' in request_data: data['mockJson'] = request_data.get('mockJson') if 'lang' in request_data: data['lang'] = request_data.get('lang') if 'method' in request_data: data['method'] = request_data.get('method') if user_data['pw'].pw_uid > 0: data['user_info'] = { 'username': user_data['pw'].pw_name, 'lve-id': user_data['pw'].pw_uid, } cli_file_path = CLOUDLINUX_USER_CLI if user_data['type'] not in [TYPE_ADMIN, TYPE_RESELLER] else CLOUDLINUX_CLI # Show unavailable page for end user if CLOUDLINUX_CLI missed if not os.path.isfile(cli_file_path): return sendError({ 'code': 503, 'context': {'pluginName': plugin_title }, 'error_id': 'ERROR.not_available_plugin', 'icon': 'disabled', 'result': ''}, 1) cli_comand = [cli_file_path, '--data={}'.format(base64_encode(json.dumps(data) .encode('utf8').strip())[0].decode('utf-8'))] (retcode, stdout, stderr) = await run_cmd_pw(user_data['pw'], cli_comand) is_privileged = user_data['type'] in [TYPE_ADMIN, TYPE_RESELLER] if stderr: logger.warning("CLI stderr for user '%s': %s", user_data['pw'].pw_name, stderr) # If decode_json is catched an exeption, send error header with backtrace try: json_data = json.loads(stdout) except: details = stdout + stderr if is_privileged else '' return sendError('ERROR.wrong_received_data', 0, 0, details) if json_data.get('result') not in ['success', 'rollback']: return sendError(json_data, 1) if stdout == '': return sendError('RESPONSE OF COMMAND IS EMPTY'); body = stdout + stderr if is_privileged else stdout return web.Response(body=body, content_type='application/json') def sendError(error_message, is_json = False, logout_signal = False, details = ''): if is_json: return web.json_response(error_message, status=503) else: response = json.dumps({ 'result': error_message, 'logoutSignal': 1 if logout_signal else 0, 'details': details }) return web.Response(status=503, body=response, content_type='application/json') def make_app(): """ Prepare web server """ app = web.Application() # add the routes app.add_routes([ web.get('/', handler_root), web.post('/login', handler_login), web.get('/app/{plugin_name}', handler_app), web.get('/logout', handler_logout), web.post('/send-request/{plugin_name}', handler_request), web.static('/assets', STATIC_FILE_PATH) ]) # set up policies policy = SessionIdentityPolicy() setup_security(app, policy, AuthorizationPolicy()) # we need to initialize aiohttp_session fernet_key = fernet.Fernet.generate_key() secret_key = base64.urlsafe_b64decode(fernet_key) setup_session(app, EncryptedCookieStorage(secret_key)) aiohttp_jinja2.setup( app, loader=jinja2.FileSystemLoader( os.path.join(SERVER_PATH, 'templates')), context_processors=[], autoescape=True, ) app.on_response_prepare.append(set_security_headers) return app def set_csrf_token(response: Response): """ Generate random csrf token and set to cookie """ response.set_cookie('csrftoken', str(uuid.uuid4())) def check_csrf_token(request: Request): """ Check csrf token """ if not request.cookies.get('csrftoken') or request.cookies.get('csrftoken') != request.headers.get('X-CSRFToken'): raise web.HTTPForbidden(body='BAD FORGERY PROTECTION TOKEN') async def set_security_headers(request, response): """ Add security headers """ if os.path.isfile(DISABLE_CSP_FLAG): return for key, value in SECURITY_HEADERS.items(): response.headers[key] = value if __name__ == '__main__': app = make_app() web.run_app( app, ssl_context=get_ssl_context(), port=get_service_port() ) PK!àS¨**lvemanager-config.jsonnu„[µü¤{"ui_config": {"inodeLimits": {"showUserInodesUsage": false}, "uiSettings": {"hideRubyApp": false, "hideLVEUserStat": false, "hidePythonApp": false, "hideNodeJsApp": false, "hidePHPextensions": false, "hideDomainsTab": false, "hidePhpApp": false, "hideXrayApp": true, "hideAccelerateWPApp": true}}}PK!~ß¡` cl.nodejs/selector.jsonnu„[µü¤{ "selector_enabled": true }PK!ª¹¦¦cl.nodejs/yum_cache.datnu„[µü¤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 PK! –$#ëëutils/cloudlinux-cli-user.pynuȯÝí#!/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() PK!…×FÑŒ-Œ-utils/cpanel_api.pynu„[µü¤# 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=` 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) PK!;¾ó1 1 utils/activatenuȯÝí#!/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 PK!-10CCutils/set_env_vars.pynuȯÝí#!/opt/cloudlinux/venv/bin/python3 -sbb # CLOS-5528 (F-14): -s skips the user-site directory so an accidental root # invocation of this script (or a compromised user home) cannot inject a # ~/.local/lib/pythonX.Y/site-packages module that shadows a stdlib name. # All shipped callers (commons/bin/{python,node,npm}_wrapper) already reject # EUID==0 and this script is not setuid, not sudoers-aliased, not called # from RPM scriptlets — the flag is defence-in-depth. -bb (warn-then-error # on implicit bytes/str mix) is preserved. Combined short flags parse # correctly under Python (-sbb == -s -b -b). # -*- 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//virtualenv// -> # /virtualenv/ 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) PK!æç-ñ  utils/node_wrappernuȯÝí#!/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" "$@"PK!¡iJêOêOutils/cloudlinux_cli_user.pynu„[µü¤# 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]: 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 PK!1N^yÂÂutils/npm_wrappernuȯÝí#!/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}" "$@" fiPK!ö=ÃÃ'Ÿ'Ÿutils/libcloudlinux.pynu„[µü¤# 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 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 = self.request_data.get("user_info") or {} if self.request_data.get("owner") == "user" and any( value is None for value in user_info.values() ): euid = os.geteuid() 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", )) 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") response = subprocess.check_output(list_to_request, shell=False, text=True) 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 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 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 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() cll.run(list_to_request) 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=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: return [ LVE_SUWRAPPER_BIN, "-n", # bypass cagefs namespace *self.enter_flags.to_lve_suwrapper_args(), str(uid), *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) PK!­88utils/python_wrappernuȯÝí#!/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}" "$@" PK!9³”{ŽŽutils/cloudlinux-selector.pynuȯÝí#!/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:])) PK!~ß¡` cl.python/selector.jsonnu„[µü¤PK!‰·y  gcl.python/yum_cache.datnu„[µü¤PK!}¾™ ±(±("Npanelless-version/daemon/server.pynuȯÝíPK!àS¨**Q*lvemanager-config.jsonnu„[µü¤PK!~ß¡` Á+cl.nodejs/selector.jsonnu„[µü¤PK!ª¹¦¦(,cl.nodejs/yum_cache.datnu„[µü¤PK! –$#ëë-utils/cloudlinux-cli-user.pynuȯÝíPK!…×FÑŒ-Œ-L/utils/cpanel_api.pynu„[µü¤PK!;¾ó1 1 ]utils/activatenuȯÝíPK!-10CCŠfutils/set_env_vars.pynuȯÝíPK!æç-ñ  yutils/node_wrappernuȯÝíPK!¡iJêOêO`zutils/cloudlinux_cli_user.pynu„[µü¤PK!1N^y–Êutils/npm_wrappernuȯÝíPK!ö=ÃÃ'Ÿ'Ÿ™Ñutils/libcloudlinux.pynu„[µü¤PK!­88qutils/python_wrappernuȯÝíPK!9³”{ŽŽ‚sutils/cloudlinux-selector.pynuȯÝíPKk\v