File manager - Edit - /usr/share/l.v.e-manager/utils/cloudlinux_cli.py
Back
# 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 # pylint: skip-file from __future__ import print_function from __future__ import division from __future__ import absolute_import import shutil import sys import json import os import re import subprocess import traceback from urllib.request import urlopen import time import logging import cldetectlib as detect from clcommon.cpapi import ( getCPName, DIRECTADMIN_NAME, admins, is_admin ) from libcagefs import CageFs from libcloudlinux import CloudlinuxCliBase from clcommon.lib.jwt_token import jwt_token_check, decode_jwt from clcommon.lib.cledition import ( is_cl_solo_edition, get_cl_edition_readable ) from clcommon.const import Feature from clcommon.cpapi import get_supported_cl_features, is_panel_feature_supported RESELLER_KERNEL_VERSION = '3.10.0-714.10.2.lve1.5.0.7' CAGEFSCTL = '/usr/sbin/cagefsctl' DONE_FLAG = '/var/lve/wizard/done.flag' FIRST_INSTALL_FLAG = '/var/lve/wizard/is_first_installation.flag' WMT_API_CLI = '/usr/share/web-monitoring-tool/wmtbin/wmt-api' SSA_CLI = '/usr/sbin/cloudlinux-ssa-manager' XRAY_CLI = '/usr/sbin/cloudlinux-xray' XRAY_AGENT_CLI = '/usr/sbin/cloudlinux-xray-manager' XRAY_USER_AGENT_CLI = '/opt/alt/php-xray/cloudlinux-xray-user-manager' XRAY_STAGING_ENABLED_FLAG = '/usr/share/alt-php-xray/staging_enabled' AWP_PREMIUM_ENABLED_FLAG = '/var/lve/enable-wpos.flag' SMART_ADVICE_CLI = '/usr/sbin/cl-smart-advice' CL_LINK_TO_CLN_CLI = '/usr/sbin/cl-link-to-cln' CL_AUTOTRACING_CLI = '/usr/sbin/cloudlinux-autotracing' CL_WIZARD_CLI = '/usr/sbin/cloudlinux-wizard' WIZARD_MODULES = { 'cagefs', 'mod_lsapi', 'mysql_governor', 'php', 'nodejs', 'python', 'ruby', } AWP_MIGRATION_NEEDED_FLAG = '/var/clwpos/admin/awp_migration_needed.v1' AWP_CLI_ADMIN = '/usr/bin/cloudlinux-awp-admin' AWP_INSTALLER_CLI = '/usr/sbin/cloudlinux-awp-installer' AWP_PLUGIN_PATH = '/usr/share/cloudlinux-awp-plugin' XRAY_PLUGIN_PATH = '/usr/share/lvemanager-xray' CL_MANAGER_UI_SETTINGS = '/opt/cpvendor/config/cl-manager-ui-settings.json' # Commands that must only ever be invoked by the server administrator. # The SPA already hides the corresponding routes/buttons behind # AdminOnlyGuard / per-service isAdmin checks, but drop_permission() # below also enforces this server-side so a reseller principal cannot # bypass the UI by posting to the backend directly. ADMIN_ONLY_COMMANDS = ( 'cloudlinux-awp-installer', 'cloudlinux-awp-admin', 'cloudlinux-xray', 'cloudlinux-xray-manager', 'cloudlinux-autotracing', 'cloudlinux-ssa-manager', 'cl-smart-advice', 'wmt-api', ) class UIFeature: """ Enum of available CloudLinux features for UI """ PHP_SELECTOR = 'php_selector' RUBY_SELECTOR = 'ruby_selector' PYTHON_SELECTOR = 'python_selector' NODEJS_SELECTOR = 'nodejs_selector' GOVERNOR = 'mysql_governor' LVE = 'lve' WIZARD = 'wizard' CAGEFS = 'cagefs' RESELLER_LIMITS = 'reseller_limits' XRAY = 'xray' ACCELERATE_WP = 'accelerate_wp' LSAPI = 'mod_lsapi' LINKING_TO_CLN = 'linking_to_cln' WEBSITE_MONITORING = 'website_monitoring' class CloudlinuxCli(CloudlinuxCliBase): resellers = None RSS_NEWS_CACHE_FILE = '/var/lve/web-team/rss_news_cache.txt' RSS_REMOTE_URL = 'https://blog.cloudlinux.com/tag/technical-blog/rss.xml' RSS_CACHE_LIFETIME = 60 IMUNIFY360_FILE = '/usr/bin/imunify360-agent' def __init__(self): super(CloudlinuxCli, self).__init__() self.command_methods.update({ 'cloudlinux-limits': self.cl_limits, 'cloudlinux-license': self.cl_license, 'cloudlinux-config' : self.cl_config, 'cloudlinux-packages': self.cl_packages, 'cloudlinux-summary': self.cl_summary, 'cloudlinux-wizard': self.cl_wizard, 'cloudlinux-support': self.cl_support, 'spa-get-admins': self.spa_get_admins, 'cagefsctl': self.cagefsctl, # cagefs 'cldiag': self.cldiag, 'spa-get-rss-news': self.spa_get_rss_news, 'spa-check-imunify-av': self.spa_check_imunify_av, 'lvectl': self.lvectl, 'cloudlinux-xray-manager': self.agent, 'cloudlinux-xray': self.cl_xray, 'external-info': self.external_info, 'cloudlinux-log': self.cl_log, 'wmt-api': self.wmt_api, 'cloudlinux-ssa-manager': self.cl_ssa, 'cloudlinux-awp-admin': self.cloudlinux_awp_admin, 'cloudlinux-autotracing': self.cl_autotracing, 'cl-smart-advice': self.cl_smart_advice, 'cl-link-to-cln': self.cl_link_to_cln, 'cloudlinux-awp-installer': self.cl_awp_installer, 'site-isolation': self.site_isolation, }) def drop_permission(self): """ Drop permission to users, if owner of script is user :return: """ data = self.request_data user_allowed_commands = ('site-isolation',) if data['owner'] not in ['admin', 'reseller']: if not (data['owner'] == 'user' and data.get('command') in user_allowed_commands): self.exit_with_error("User not allowed") # Admin-only command gate: even though 'reseller' passes the # ['admin', 'reseller'] check above, certain server-wide maintenance # commands must be limited to the server administrator. This mirrors # the SPA-side AdminOnlyGuard # and per-service isAdmin checks so a reseller principal cannot # bypass the UI by posting to the backend directly. if data['owner'] != 'admin' and data.get('command') in ADMIN_ONLY_COMMANDS: self.exit_with_error("User not allowed") super(CloudlinuxCli, self).drop_permission() if data['owner'] == 'reseller' and data.get('command') != 'external-info': if self.is_user_in_admin_list(): self.exit_with_error( code=503, error_id='ERROR.login_by_admin', icon='info') if self.get_lve_version() <= 9: if self.get_cloudlinux_version() in ['el7', 'el6h']: self.exit_with_error( code=503, error_id='ERROR.not_supported_kernel', context={'resellerKernelVersion': RESELLER_KERNEL_VERSION}, icon='info') else: self.exit_with_error( code=503, error_id='ERROR.not_supported_OS', context={'resellerKernelVersion': RESELLER_KERNEL_VERSION}, icon='info') if not self.is_activated_reseller(): self.exit_with_error( code=503, error_id='ERROR.not_available_plugin', context={'pluginName': 'CloudLinux Manager'}, icon='disabled') if os.getuid() != 0: os.setgroups([]) os.setgid(0) os.setuid(0) def cl_summary(self): list_to_request = self.prepair_params_for_command() try: output = subprocess.check_output(['/usr/sbin/cloudlinux-summary'] + list_to_request, stderr=subprocess.STDOUT, shell=False, text=True) except subprocess.CalledProcessError as processError: output = processError.output try: result = json.loads(output) result['statistic_enabled'] = self._is_statistics_enabled_by_admin() result['ignore'] = True print(json.dumps(result)) except: print(output) sys.exit(0) @staticmethod def _has_admin_only_limit_param(params, admin_only): # docopt (the cloudlinux-limits parser) accepts unique-prefix abbreviations, so # --mysql-g resolves to --mysql-gov. Match a param key if it equals OR is a prefix # of any admin-only canonical option, so an abbreviated key cannot evade the gate. # No legitimate non-admin limit key is a prefix of an admin-only option, so this # does not over-block. (CLOS-4594/F-51) return any( canon == key or canon.startswith(key) for key in params for canon in admin_only ) def cl_limits(self): # MySQL Governor / per-user MySQL limit mutations (plus cagefs, inodes) are # admin-only. The caller role is known here (request owner); cloudlinux-limits # runs as root and has no caller identity of its own (its can_get_governor_limits() # only checks Governor state, not who is calling). Enforce it here, sourcing the # admin-only set from cloudlinux-limits' own arg parser (ADMIN_ONLY_OPTIONS) so the # dispatcher gate can never drift from the backend. (CLOS-4594/F-51) if self.request_data.get("owner") != "admin": from cllimits.lib.arg_parsers import ADMIN_ONLY_OPTIONS admin_only = frozenset(opt.lstrip("-") for opt in ADMIN_ONLY_OPTIONS) params = self.request_data.get("params") or {} if self._has_admin_only_limit_param(params, admin_only): self.exit_with_error("Command is not available") from cllimits.lib.limits import CloudlinuxLimits list_to_request = self.prepair_params_for_command() cl = CloudlinuxLimits() cl.run(list_to_request) def cl_license(self): from cllicense.license import CloudlinuxLicense if self.request_data.get('params', {}).get('key') == '': self.request_data['params']['ip'] = self.request_data['params'].pop('key') list_to_request = self.prepair_params_for_command() cll = CloudlinuxLicense() cll.run(list_to_request) def cl_config(self): from clconfig.config import ClConfig """ Main run function """ list_to_request = self.prepair_params_for_command() cll = ClConfig() cll.run(list_to_request) def cl_packages(self): from clpackages.packages import CloudlinuxPackages # TODO: fix cloudlinux-packages and remove this ugly unicode->str->unicode->str transformation list_to_request = self.prepair_params_for_command(escaped_strings=True) cll = CloudlinuxPackages() cll.run(list_to_request) def spa_get_admins(self): if self.request_data['owner'] == 'admin' and getCPName() == DIRECTADMIN_NAME: print(json.dumps( {"result":"success", "list": list(admins())})) sys.exit(0) else: self.exit_with_error('Command is not available') def spa_get_rss_news(self): content = None cur_time = time.time() last_modified_time = os.path.getmtime(self.RSS_NEWS_CACHE_FILE) \ if os.path.exists(self.RSS_NEWS_CACHE_FILE) else 0 age_of_file = (cur_time - last_modified_time) / 60 # in minutes if age_of_file > self.RSS_CACHE_LIFETIME: # for file which older than hour or it is just created self._update_rss_news() try: with open(self.RSS_NEWS_CACHE_FILE, 'r') as f: content = f.read() except IOError: pass print(json.dumps({ "result":"success", "content": content, "age" : age_of_file })) sys.exit(0) def _update_rss_news(self): if not os.path.exists(os.path.dirname(self.RSS_NEWS_CACHE_FILE)): os.makedirs(os.path.dirname(self.RSS_NEWS_CACHE_FILE)) try: xmlf = urlopen(self.RSS_REMOTE_URL, timeout=10) with open(self.RSS_NEWS_CACHE_FILE, 'w+') as f: f.write(xmlf.read().decode('utf-8')) except IOError: pass def spa_check_imunify_av(self): result = 'installed' if os.path.exists(self.IMUNIFY360_FILE) else 'not_installed' print(json.dumps({"result": "success", "response": result })) def cagefsctl(self): if self.request_data['owner'] != 'admin': return cagefs = CageFs() method = self.request_data.get('method') if method: func = { 'status': cagefs.status, 'init': cagefs.init, 'update': cagefs.update, 'enable': lambda: cagefs.change_status('enable', self.request_data['params'].get('users')), 'disable': lambda: cagefs.change_status('disable', self.request_data['params'].get('users')), 'log': lambda: cagefs.get_log(self.request_data['params'].get('operation')), 'download-log': lambda: cagefs.download_log(self.request_data['params'].get('operation')), 'log_data': lambda: cagefs.cagefs_log_data(self.request_data['params'].get('operation')), }.get(method) print(json.dumps({"result": "success", "response": func()})) else: list_to_request = self.prepair_params_for_command(with_json=False) response = cagefs.run(list_to_request) print(json.dumps({"result": "success", "response": response})) sys.exit(0) def _run_cagefsctl(self, *args): p = subprocess.run( [CAGEFSCTL] + list(args), capture_output=True, text=True, ) if p.returncode != 0: logging.error( 'cagefsctl %s failed (rc=%s): %s', args, p.returncode, p.stderr.strip() or p.stdout.strip(), ) self.exit_with_error('Operation failed') def site_isolation(self): owner = self.request_data['owner'] method = self.request_data.get('method') params = self.request_data.get('params', {}) or {} if owner == 'admin': self._site_isolation_admin(method, params) elif owner == 'user': username = self.user_info.get('username') if not username: self.exit_with_error('User info not available') self._site_isolation_user(method, params, username) else: self.exit_with_error('Not allowed') def _site_isolation_admin(self, method, params): if method == 'get-status': from clcagefslib.domain import ( is_website_isolation_feature_available, is_website_isolation_allowed_server_wide, get_isolation_user_mode, ) server_wide = is_website_isolation_allowed_server_wide() self.exit_with_success({ 'featureAvailable': is_website_isolation_feature_available(), 'serverWideAllowed': server_wide, 'userMode': get_isolation_user_mode() if server_wide else None, }) elif method == 'get-user-status': from clcagefslib.domain import is_website_isolation_allowed_for_user user = params.get('user') if not user: self.exit_with_error('Missing user parameter') self.exit_with_success({ 'allowed': is_website_isolation_allowed_for_user(user), }) elif method in ('allow', 'deny'): users = self._parse_user_list(params) if not users: self.exit_with_error('Missing user or users parameter') self._run_cagefsctl('--site-isolation-' + method, *users) self.exit_with_success({'users': users}) elif method == 'allow-all': self._run_cagefsctl('--site-isolation-allow-all') self.exit_with_success() elif method == 'deny-all': self._run_cagefsctl('--site-isolation-deny-all') self.exit_with_success() elif method == 'list-domains': from clcagefslib.domain import users_with_enabled_domain_isolation self.exit_with_success({ 'domains': users_with_enabled_domain_isolation(), }) elif method == 'list-users': from clcagefslib.domain import ( is_website_isolation_allowed_for_user, get_websites_with_enabled_isolation, ) from clcommon.cpapi import cpusers result = [] for u in cpusers(): allowed = is_website_isolation_allowed_for_user(u) result.append({ 'username': u, 'isolationAllowed': allowed, 'isolatedDomains': ( get_websites_with_enabled_isolation(u) if allowed else [] ), }) self.exit_with_success({'users': result}) elif method == 'list-all-domains': self._site_isolation_list_all_domains() elif method == 'get-domain-versions': self._site_isolation_domain_versions(params) else: self.exit_with_error('Unknown method: ' + str(method)) def _site_isolation_user(self, method, params, username): if method == 'get-status': from clcagefslib.domain import ( is_website_isolation_feature_available, is_website_isolation_allowed_server_wide, is_website_isolation_allowed_for_user, get_websites_with_enabled_isolation, ) from clcommon.cpapi import userdomains from clselect.clselectdomains import ( get_all_selector_compatible_domains_flat, ) 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 ) isolated = ( set(get_websites_with_enabled_isolation(username)) if user_allowed else set() ) selector_domains = get_all_selector_compatible_domains_flat() domains_info = [ { 'domain': d, 'isolated': d in isolated, 'selectorCompatible': d in selector_domains, } for d, _ in userdomains(username) ] self.exit_with_success({ 'featureAvailable': feature_available, 'allowed': user_allowed, 'domains': domains_info, }) elif method in ('enable', 'disable'): domain = params.get('domain') if not domain: self.exit_with_error('Missing domain parameter') # Mirror the admin-deny gate from get-status above: a user the # admin marked as not allowed must not be able to toggle # isolation on their own domain by POSTing directly. try: from clcagefslib.domain import ( is_website_isolation_allowed_server_wide, is_website_isolation_allowed_for_user, ) 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. Mirrors the # defensive shape now in cloudlinux_cli_user._site_isolation_toggle. self.exit_with_error('Site isolation feature is unavailable') from clcommon.cpapi import userdomains user_domains = [d for d, _ in userdomains(username)] if domain not in user_domains: self.exit_with_error('Domain does not belong to user') self._run_cagefsctl('--site-isolation-' + method, domain) self.exit_with_success({'domain': domain}) elif method == 'get-domain-versions': self._site_isolation_domain_versions(params, user=username) else: self.exit_with_error('Unknown method: ' + str(method)) def _site_isolation_domain_versions(self, params, user=None): from clcagefslib.domain import get_websites_with_enabled_isolation from clcommon.cpapi import cpusers from clselect import ClUserSelect user_selector = ClUserSelect('php') users_to_check = [user] if user else list(cpusers()) domain_versions = {} for u in users_to_check: user_versions = {} try: isolated = get_websites_with_enabled_isolation(u) for domain_name in isolated: try: ver_info = user_selector.get_version(u, domain_name) if ver_info and ver_info[0]: user_versions[domain_name] = ver_info[0] except Exception: logging.debug( 'get-domain-versions: failed to get version ' 'for domain %s of user %s', domain_name, u, exc_info=True, ) except Exception: logging.debug( 'get-domain-versions: failed to list domains for user %s', u, exc_info=True, ) domain_versions[u] = user_versions self.exit_with_success({'domainVersions': domain_versions}) def _site_isolation_list_all_domains(self): """ Return all domains grouped by user with handler info, isolation status, and selector availability. Used by Plesk/DA/vendor panels where CpanelRepo is unavailable. Uses public panel-agnostic APIs: - cpapi.get_domains_php_info() for domain/handler/version (works for cPanel, Plesk, DA, and vendor panels) - clselectstatistics.get_php_selector_compatible_domains() for per-domain selector compatibility (handler + panel rules) - clcagefslib for isolation status """ from clcommon.cpapi import get_domains_php_info from clselect.clselectdomains import ( get_all_selector_compatible_domains_flat, ) from clcagefslib.domain import ( is_website_isolation_allowed_for_user, get_websites_with_enabled_isolation, ) domains_php_info = get_domains_php_info() selector_domains = get_all_selector_compatible_domains_flat() users = {} for domain, info in domains_php_info.items(): username = info.get('username', '') if not username: continue if username not in users: allowed = False isolated_domains = [] try: allowed = is_website_isolation_allowed_for_user(username) if allowed: isolated_domains = ( get_websites_with_enabled_isolation(username) ) except Exception: logging.debug( 'list-all-domains: failed for user %s', username, exc_info=True, ) users[username] = { 'isolationAllowed': allowed, 'isolatedDomains': set(isolated_domains), 'domains': [], } isolated_set = users[username]['isolatedDomains'] handler = info.get('handler_type', '') users[username]['domains'].append({ 'domain': domain, 'handlerType': handler or '', 'phpVersion': info.get('php_version_id', ''), 'selectorCompatible': domain in selector_domains, 'isolated': domain in isolated_set, }) result = {} for username, data in users.items(): result[username] = { 'isolationAllowed': data['isolationAllowed'], 'domains': data['domains'], } self.exit_with_success({'users': result}) @staticmethod def _parse_user_list(params): users_str = params.get('users', '') user = params.get('user', '') return users_str.split() if users_str else ([user] if user else []) def cldiag(self): # cldiag aggregates server-wide diagnostic output (suexec/suphp/php-cfg/ # cagefs/mod_lsapi checkers, etc.) which is not scoped to a single # reseller's tenancy. Match the admin-only gate used by the sibling # cl_support and cl_wizard handlers above to keep reseller principals # out of an admin-scoped information surface. if self.request_data['owner'] != 'admin': return params = ['/usr/bin/cldiag'] + self.prepair_params_for_command() if len(params) <= 2: params.append('--all') result = None try: with open(os.devnull, 'w') as devnull: output = subprocess.check_output(params, stderr=devnull, shell=False, text=True) result = json.loads(output) except subprocess.CalledProcessError as e: try: result = json.loads(e.output) except: result = e.output except OSError as e: self.exit_with_error('Can\'t call cldiag: ' + str(e)) if result: print(json.dumps({ 'result': 'success', 'data': result})) sys.exit(0) def cl_support(self): if self.request_data['owner'] != 'admin': return method = self.request_data.get('method') params = self.request_data.get('params', {}) or {} if method == 'custom-fields': from libsupport import build_custom_fields_from_send_params fields = build_custom_fields_from_send_params(params) self.exit_with_success({ "fields": fields }) else: self.exit_with_error('Method is not available') def cl_wizard(self): if self.request_data['owner'] != 'admin': return method = self.request_data.get('method') if not method: self.exit_with_error('Module unavailable', ignore_errors=True) elif method == 'status' and 'initial' not in self.request_data.get('params', {}): self.wizard_get_status() elif method == 'status' and 'initial' in self.request_data.get('params', {}): self.wizard_get_initial() elif method == 'log': self.wizard_log() elif method == 'log_data': self.wizard_log_data() elif method == 'finish': self.wizard_finish() elif method == 'cancel': self.wizard_cancel() else: list_to_request = self.prepair_params_for_command(with_json=False) try: print(self.run_util(CL_WIZARD_CLI, *list_to_request)) except Exception as e: self.exit_with_error(traceback.format_exc()) sys.exit(0) def spa_detect_first_installation(self): print(json.dumps({ "result":"success", "is_first_installation": os.path.exists(FIRST_INSTALL_FLAG) })) sys.exit(0) def is_activated_reseller(self): username = self.user_info['username'] if self.resellers is None: p = subprocess.Popen( ['lvectl', 'list-reseller', '--with-name', '--json'], stdout=subprocess.PIPE, stderr=subprocess.PIPE) (res_in_json, err) = p.communicate() if p.returncode != 0: self.check_license() try: self.resellers = [reseller.get('ID', '').split(':')[-1] \ for reseller in json.loads(res_in_json)['data']] except: self.resellers = [] return username in self.resellers def is_user_in_admin_list(self): """ Check user in admin list (disable reseller plugin for admin in DA) """ if getCPName() == DIRECTADMIN_NAME: if is_admin(self.user_info['username']): return True return False def _is_statistics_enabled_by_admin(self): # FIXME: move to cllib return detect.get_boolean_param(detect.CL_CONFIG_FILE, 'cl_statistics_enabled') def checkIsFirstInstall(self): return os.path.exists(FIRST_INSTALL_FLAG) def wizard_get_status(self): if not self.checkIsFirstInstall() or os.path.exists(DONE_FLAG): self.exit_with_success({'wizard_status': 'finish'}) else: try: result = self.run_util(CL_WIZARD_CLI, 'status', ignore_errors=True) data = json.loads(result) if data.get('wizard_status') == 'idle': self.exit_with_success({'wizard_status': 'first_install'}) else: print(result) except ValueError as e: print(result) except Exception as e: self.exit_with_error(traceback.format_exc()) def wizard_get_initial(self): result = self.run_util(CL_WIZARD_CLI, 'status', '--initial', ignore_errors=True) data = json.loads(result) if 'RPM DB check error' in data.get('result'): self.exit_with_error('RPM DB is Corrupted', context={ 'message': data.get('result'), }) else: print(result) sys.exit(0) def wizard_log(self): """ Check log file exists and return filename to wrapper to seed """ filepath = self._get_wizard_log_file() print(json.dumps({ "result": "file", "filepath": filepath, "filesize": os.path.getsize(filepath) })) def wizard_log_data(self): """ Check log file exists and return filename to wrapper to seed """ filepath = self._get_wizard_log_file() with open(filepath) as log_file: shutil.copyfileobj(log_file, sys.stdout, 20*1024) sys.exit(0) def _get_wizard_log_file(self): module_name = self._get_wizard_module_name(allow_crash=True) data = self.run_util(CL_WIZARD_CLI, 'status') data = json.loads(data) if module_name == 'crash': filepath = data.get('crash_log') else: filepath = next( (module.get('log_file') for module in data.get('modules', []) if module.get('name') == module_name ), None) if not filepath: self.exit_with_error("Log file {} not found".format(filepath)) if not os.path.isfile(filepath): self.exit_with_error("File {} does not exist".format(filepath)) if not os.access(filepath, os.R_OK): self.exit_with_error("File {} not available for reading".format(filepath)) return filepath def _get_wizard_module_name(self, allow_crash=False): module_name = self.request_data.get('params', {}).get('module') if allow_crash and module_name == 'crash': return module_name if module_name not in WIZARD_MODULES: self.exit_with_error('Invalid wizard module') return module_name def wizard_cancel(self): module_name = self._get_wizard_module_name() print(self.run_util(CL_WIZARD_CLI, 'cancel', module_name)) sys.exit(0) def wizard_finish(self): if os.path.exists(FIRST_INSTALL_FLAG): os.remove(FIRST_INSTALL_FLAG) print(self.run_util(CL_WIZARD_CLI, 'finish', ignore_errors=True)) sys.exit(0) def lvectl(self): """run lvectl command with arguments""" POSITION_ARGUMENTS_LIST = ['lveid', 'username', 'package', 'pid'] REPLACE_METHODS = { 'apply-all': 'apply all', 'destroy-all': 'destroy all', } position_arguments = [] params = self.request_data.get('params', {}) if params.get('lveid') is not None and params.get('username') is not None: print(json.dumps({'error': 'lveid and username should not be presented together'})) exit(0) if self.request_data.get('method') in REPLACE_METHODS: self.request_data['method'] = REPLACE_METHODS.get(self.request_data['method']) for argument in POSITION_ARGUMENTS_LIST: value = params.pop(argument, None) if value: value = str(value) if ' ' in value or value.startswith('-'): self.exit_with_error("BAD REQUEST 3") position_arguments.append(value) if position_arguments: self.request_data['method'] = ' '.join([self.request_data.get('method', '')] + position_arguments) stdin = params.pop('stdin', None) list_to_request = self.prepair_params_for_command() try: print(self.run_util('/usr/sbin/lvectl', *list_to_request, stdin=stdin)) except Exception as e: self.exit_with_error(traceback.format_exc()) def agent(self): """ Agnet requests :return: """ def xray_error_checker(json_result): return json_result.get('status') != 'ok' list_to_request = self.prepair_params_for_command(with_json=False) try: response = self.run_util(XRAY_AGENT_CLI, *list_to_request, error_checker=xray_error_checker) response = json.loads(response) self.exit_with_success(response) except Exception as e: self.exit_with_error(traceback.format_exc()) def cl_autotracing(self): """ Run cloudlinux-autotracing :return: Any """ list_to_request = self.prepair_params_for_command(with_json=False) try: response = self.run_util(CL_AUTOTRACING_CLI, *list_to_request) response = json.loads(response) self.exit_with_success(response) except Exception as e: self.exit_with_error(traceback.format_exc()) @staticmethod def compare_versions(ver1, ver2): """ :param ver1: string :param ver2: string :return: True if first version is bigger or equal to the second """ ver1_tuple = tuple(map(int, ver1.split("."))) ver2_tuple = tuple(map(int, ver2.split("."))) return ver1_tuple >= ver2_tuple @staticmethod def get_customer_id_from_jwt(jwt): """ Return the non-secret CLN customer id from server-side JWT claims. """ try: if not jwt: return None customer_id = int(decode_jwt(jwt).get('client_id')) return customer_id if customer_id > 0 else None except Exception: logging.debug('Unable to read customer id from JWT', exc_info=True) return None def merge_dicts(self, dict1, dict2): """ Merge two dicts and return merge result """ result = dict1.copy() result.update(dict2) return result def read_json_config_file(self, config_path): """ Read json file and return result """ try: with open(config_path, 'r') as config_file: return json.load(config_file) except FileNotFoundError: return {} def cl_xray(self): """ Check status and install xray utility Return status and server_id """ list_to_request = self.prepair_params_for_command(with_json=False) util_response = json.loads(self.run_util(XRAY_CLI, *list_to_request)) if util_response.get('result') == 'success': self.exit_with_success(util_response) else: self.exit_with_error(util_response.get('response')) def get_available_cl_features(self): supported_features = get_supported_cl_features() # Rewrite supported features for UI with custom config if os.path.isfile(CL_MANAGER_UI_SETTINGS): config = self.read_json_config_file(CL_MANAGER_UI_SETTINGS) features_from_config = config.get("features_override", {}) return self.merge_dicts(supported_features, features_from_config) return supported_features def get_supported_features_for_ui(self): features = self.get_available_cl_features() return { UIFeature.PHP_SELECTOR: features.get(Feature.PHP_SELECTOR, False), # TODO: Should be replaced to is_php_supported() UIFeature.RUBY_SELECTOR: features.get(Feature.RUBY_SELECTOR, False), UIFeature.PYTHON_SELECTOR: features.get(Feature.PYTHON_SELECTOR, False), UIFeature.NODEJS_SELECTOR: features.get(Feature.NODEJS_SELECTOR, False), UIFeature.GOVERNOR: self.is_governor_available(features), UIFeature.CAGEFS: features.get(Feature.CAGEFS, False), UIFeature.LVE: features.get(Feature.LVE, False), UIFeature.RESELLER_LIMITS: features.get(Feature.RESELLER_LIMITS, False), UIFeature.XRAY: features.get(Feature.XRAY, False), UIFeature.ACCELERATE_WP: features.get(Feature.WPOS, False), UIFeature.WIZARD: features.get(Feature.WIZARD, False), UIFeature.LSAPI: features.get(Feature.LSAPI, False), UIFeature.LINKING_TO_CLN: features.get(UIFeature.LINKING_TO_CLN, True), UIFeature.WEBSITE_MONITORING: features.get(UIFeature.WEBSITE_MONITORING, True), } def is_ssa_installed(self): """ Check if Slow Site Analyzer availability """ ssa_binary_present = os.path.exists(SSA_CLI) return ssa_binary_present def is_wmt_installed(self): """ Check if Website Monitoring availability """ return os.path.exists(WMT_API_CLI) def is_smart_advice_installed(self): """ Check Smart Advice availability """ smart_advice_binary_present = os.path.exists(SMART_ADVICE_CLI) return smart_advice_binary_present def is_xray_installed(self): """ Check X-Ray availability """ xray_binary_present = all([os.path.isfile(XRAY_AGENT_CLI) and os.path.isfile(XRAY_USER_AGENT_CLI)]) return xray_binary_present def is_awp_installed(self): """ Check AccelerateWP availability """ awp_plugin_installed = os.path.exists(AWP_PLUGIN_PATH) return awp_plugin_installed def is_cagefs_installed(self): return os.path.isfile(CAGEFSCTL) and os.path.isdir('/usr/share/cagefs-skeleton/bin') def is_site_isolation_available(self): if not self.is_cagefs_installed(): return False try: from clcagefslib.domain import is_website_isolation_feature_available return is_website_isolation_feature_available() except Exception: return False def get_site_isolation_unavailable_reason(self): """ When isolation is not available, determine why so the frontend can show an appropriate warning. """ if not self.is_cagefs_installed(): return None try: from clcagefslib.domain import is_website_isolation_feature_available if is_website_isolation_feature_available(): return None import platform release = platform.release() if '.el7' in release: return 'cl7_unsupported' return 'unsupported_platform' except Exception: logging.debug( 'get_site_isolation_unavailable_reason failed', exc_info=True, ) return None def is_governor_available(self, features): """ Check if Governor available in UI """ is_governor = features.get(Feature.GOVERNOR, False) is_lve = features.get(Feature.LVE, False) return all([is_governor, is_lve]) def external_info(self): """ Return external info based on the owner type (admin or reseller). """ if self.request_data['owner'] == 'admin': self._get_admin_external_info() elif self.request_data['owner'] == 'reseller': self._get_reseller_external_info() else: self.exit_with_error('User not allowed') def _get_admin_external_info(self): """ Gather and return external info needed for admin. """ from clcommon.utils import get_rhn_systemid_value system_id = get_rhn_systemid_value('system_id') or '' if system_id: system_id = system_id.split('-')[1] is_valid, _, jwt = jwt_token_check() customer_id = self.get_customer_id_from_jwt(jwt) self.exit_with_success({ 'system_id': system_id, 'customer_id': customer_id, 'cl_plus': is_valid, 'is_cl_solo_edition': is_cl_solo_edition(skip_jwt_check=True), 'xray_staging_enabled': os.path.isfile(XRAY_STAGING_ENABLED_FLAG), 'awp_premium': os.path.isfile(AWP_PREMIUM_ENABLED_FLAG), 'awp_plugin_version': self.get_awp_plugin_version(), 'xray_plugin_version': self.get_xray_plugin_version(), 'plugin_installed': True, 'autotracing': os.path.isfile(CL_AUTOTRACING_CLI), 'awp_migration_needed': os.path.isfile(AWP_MIGRATION_NEEDED_FLAG), 'cl_edition': get_cl_edition_readable(), 'supported_cl_features': self.get_supported_features_for_ui(), 'xray_installed': self.is_xray_installed(), 'accelerate_wp_installed': self.is_awp_installed(), 'ssa_installed': self.is_ssa_installed(), 'wmt_installed': self.is_wmt_installed(), 'awp_installed': self.is_awp_installed(), 'smart_advice_installed': self.is_smart_advice_installed(), 'cagefs_installed': self.is_cagefs_installed(), 'site_isolation_available': self.is_site_isolation_available(), 'site_isolation_unavailable_reason': self.get_site_isolation_unavailable_reason(), 'server_ip': self.get_server_ip(), }) def _get_reseller_external_info(self): """ Gather and return external info needed for reseller. """ self.exit_with_success({ 'is_cl_solo_edition': is_cl_solo_edition(skip_jwt_check=True), 'cl_edition': get_cl_edition_readable(), 'supported_cl_features': self.get_supported_features_for_ui(), 'server_ip': self.get_server_ip(), }) def wmt_api(self): list_to_request = self.prepair_params_for_command(with_json=False) util_response = json.loads(self.run_util(WMT_API_CLI, *list_to_request)) if util_response.get('result') == 'success': self.exit_with_success(util_response) else: self.exit_with_error(util_response.get('response')) def cl_ssa(self): list_to_request = self.prepair_params_for_command(with_json=False) util_response = json.loads(self.run_util(SSA_CLI, *list_to_request)) if util_response.get('result') == 'success': self.exit_with_success(util_response) else: self.exit_with_error(util_response.get('response')) # Control characters other than '\t' (TAB). The global input filter in # libcloudlinux.check_param_value() only rejects '\n' plus shell # metacharacters, so '\r', '\x1b' (ESC / ANSI), '\x08' (BS), and the # rest of C0/C1 reach this sink unmodified. Resellers can dispatch # 'cloudlinux-log' (not in ADMIN_ONLY_COMMANDS), so without scrubbing # they can smear lines in the root-owned audit log via '\r' or inject # ANSI escape sequences that confuse terminal-based log viewers. _CL_LOG_CONTROL_CHARS_RE = re.compile(r'[\x00-\x08\x0a-\x1f\x7f-\x9f]') def cl_log(self): """ Writes log message to the log file :return: """ message = self.request_data.get('params', {}).get('message') if message: if not isinstance(message, str): message = str(message) # Replace control chars (CR, ESC, BS, ...) with '?' so a reseller # principal cannot manipulate terminal-based viewers of the # root-owned audit log. '\n' is already blocked upstream by # check_param_value(); '\t' is preserved for readability. message = self._CL_LOG_CONTROL_CHARS_RE.sub('?', message) logging.basicConfig(filename='/var/log/cloudlinux/lvemanager.log', level=logging.INFO, format='%(asctime)s %(message)s') logging.info(message) self.exit_with_success() def cloudlinux_awp_admin(self): """ Run cloudlinux-awp-admin utility with args """ list_to_request = self.prepair_params_for_command(with_json=False) util_response = json.loads(self.run_util(AWP_CLI_ADMIN, *list_to_request)) if util_response.get('result') == 'success': self.exit_with_success(util_response) else: self.exit_with_error(util_response.get('response')) def cl_smart_advice(self): 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([SMART_ADVICE_CLI, *list_to_request], stdin=subprocess.PIPE,stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) self.exit_with_success() util_response = json.loads(self.run_util(SMART_ADVICE_CLI, *list_to_request)) if util_response.get('result') == 'success': self.exit_with_success(util_response) else: self.exit_with_error(util_response.get('response')) def cl_link_to_cln(self): # Linking the server to CLN is a server-wide admin operation: the # underlying /usr/sbin/cl-link-to-cln binary is invoked with the # caller-supplied token after drop_permission has already restored # UID 0, so there is no in-band way for the binary to distinguish # reseller-originated calls from admin ones. Gate here. if self.request_data['owner'] != 'admin': self.exit_with_error("User not allowed") list_to_request = self.prepair_params_for_command(with_json=False) util_response = json.loads(self.run_util(CL_LINK_TO_CLN_CLI, *list_to_request)) if util_response.get('result') == 'success': self.exit_with_success(util_response) else: self.exit_with_error(util_response.get('response')) def cl_awp_installer(self): """ Run commands for AccelerateWP """ # Admin-only: installer triggers server-global state (RPM install of # cloudlinux-awp-plugin, /var/lve/clflags/enable_awp_all_servers.flag). # Reseller is admitted through drop_permission for other commands; # gate AWP installer explicitly, matching the cagefsctl pattern above. if self.request_data['owner'] != 'admin': return list_to_request = self.prepair_params_for_command(with_json=False) util_response = json.loads(self.run_util(AWP_INSTALLER_CLI, *list_to_request)) if util_response.get('result') == 'success': self.exit_with_success(util_response) else: self.exit_with_error(util_response.get('response')) def get_awp_plugin_version(self): """ Get version of the AccelerateWP plugin """ try: version_file = AWP_PLUGIN_PATH + '/version' with open(version_file, 'r') as f: return f.read().strip() except: return None def get_xray_plugin_version(self): """ Get version of the X-Ray plugin """ try: version_file = XRAY_PLUGIN_PATH + '/version' with open(version_file, 'r') as f: return f.read().strip() except: return None
| ver. 1.4 |
Github
|
.
| PHP 8.1.34 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings