Файловый менеджер

Файловый менеджер - Редактировать - /opt/alt/php-xray/php/smart-advice-plugin/run.py

Назад
#!/opt/cloudlinux/venv/bin/python3 -Ibb ## # Copyright (с) Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://www.cloudlinux.com/legal/ ## import grp import hashlib import logging import os import re import pwd import stat import subprocess import sys import time import json import shutil from dataclasses import dataclass from enum import Enum, auto from packaging.version import ( InvalidVersion, Version, parse as parse_version ) from xray.internal.utils import sentry_init from xray.internal.user_plugin_utils import check_for_root from xray.analytics import report_analytics from xray.imunify_manager import ADV_TYPE as IMUNIFY_ADV_TYPE from clcommon.cpapi import getCPName from clcommon.cpapi.pluginlib import getuser class Action(Enum): INSTALL = auto() UNINSTALL = auto() @dataclass class Args: path_to_php: str path_to_wp: str path_to_mu_plugin: str advice_list_json: str emails: str login_url: str action: Action = Action.INSTALL @dataclass class WebsiteInfo: php_version: Version wp_version: Version wp_must_use_plugins_path: str def _get_wp_cli_environment(disable_smart_advice: bool = False): env = dict(SKIP_PLUGINS_LOADING='1') if disable_smart_advice: env.update(dict(CL_SMART_ADVICE_DISABLED='1')) return env def get_wp_cli_command(args: Args): return ["/opt/clwpos/wp-cli-wrapped", args.path_to_php, args.path_to_wp] def _gather_must_use_plugin_dir(args): try: must_use_plugin_dir = subprocess.check_output( [get_wp_cli_command(args), 'eval', 'echo WPMU_PLUGIN_DIR;'], text=True, env=_get_wp_cli_environment(disable_smart_advice=True), stderr=subprocess.PIPE).strip() except subprocess.CalledProcessError as e: logging.error('Unable to read WPMU_PLUGIN_DIR constant; ' '\nstdout=`%s`\nstderr=`%s`,\nreturncode=%s' '', e.stdout.strip('\n'), e.stderr.strip('\n'), e.returncode) must_use_plugin_dir = None if must_use_plugin_dir: return must_use_plugin_dir try: wp_content_dir = subprocess.check_output( [get_wp_cli_command(args), 'eval', 'echo WP_CONTENT_DIR;'], text=True, env=_get_wp_cli_environment(disable_smart_advice=True), stderr=subprocess.PIPE).strip() except subprocess.CalledProcessError as e: logging.error(f"Error reading WordPress constant WP_CONTENT_DIR.; " f"\nstdout=`%s`\nstderr=`%s`,\nreturncode=%s" '', e.stdout.strip('\n'), e.stderr.strip('\n'), e.returncode) return None else: if not wp_content_dir: logging.error("The path to the folder with must use plugins is empty.") return None must_use_plugin_dir = os.path.join(wp_content_dir, 'mu-plugins') return must_use_plugin_dir def _gather_advice_ids(args: Args) -> str: try: advice_ids = [] advice_dict = json.loads(args.advice_list_json) for advice in advice_dict['data']: if advice['advice']['type'] == IMUNIFY_ADV_TYPE: logging.info('Advice has IMUNIFY TYPE, skip: %s', str(advice)) continue advice_ids.append(advice['advice']['id']) advice_row = ','.join(str(id) for id in advice_ids) except (KeyError, ValueError, TypeError): logging.error('[Analytics] Unable to get advice list.') return None return advice_row def _user_hash(username): # when > 1 user return ','.join([hashlib.md5(u.encode("utf-8")).hexdigest() for u in username.split(',')]) def prepare_system_analytics_data(user_name: str, user_hash=None, journey_id=None) -> str: return json.dumps({ 'user_hash': user_hash or _user_hash(user_name), 'journey_id': journey_id or 'system', 'username': user_name }) def _report_sync_or_error(args: Args, event: str) -> None: """ Report advice sync status """ username = getuser() if username is None: return analytics_data = prepare_system_analytics_data(username) if analytics_data is None: return advice_id = _gather_advice_ids(args) if not advice_id: return source = getCPName() if source is None: return report_analytics(analytics_data, advice_id, source.lower(), event) def _gather_wordpress_version(args): try: wordpress_version_raw = subprocess.check_output( [get_wp_cli_command(args), 'core', 'version'], text=True, env=_get_wp_cli_environment(disable_smart_advice=True), stderr=subprocess.PIPE).strip() except subprocess.CalledProcessError as e: logging.error(f"Error happened while reading WordPress versions." f"\nstdout=`%s`\nstderr=`%s`,\nreturncode=%s" '', e.stdout.strip('\n'), e.stderr.strip('\n'), e.returncode) return None try: wordpress_version = parse_version(wordpress_version_raw) except InvalidVersion: logging.error("WordPress version %s is unparsable.", wordpress_version_raw) return None if wordpress_version < parse_version('4.6'): logging.error("WordPress version %s is not supported.", wordpress_version_raw) return None return wordpress_version def _gather_php_version(args: Args): try: php_version_raw = subprocess.check_output( [args.path_to_php, '-r', 'echo PHP_VERSION;'], stderr=subprocess.PIPE, text=True, env=_get_wp_cli_environment(), ).strip() except subprocess.CalledProcessError as e: logging.error('Unable to get php version' f"\nstdout=`%s`\nstderr=`%s`,\nreturncode=%s" '', e.stdout.strip('\n'), e.stderr.strip('\n'), e.returncode) return None try: php_version = parse_version(php_version_raw) except InvalidVersion: logging.error("PHP version %s cannot be parsed.", php_version_raw) return None if php_version < parse_version('5.6'): logging.error("PHP version %s is not supported.", php_version) return None return php_version def gather_information(args: Args): if (php_version := _gather_php_version(args)) is None: return None if (wordpress_version := _gather_wordpress_version(args)) is None: return None return WebsiteInfo( php_version=php_version, wp_version=wordpress_version, wp_must_use_plugins_path=args.path_to_mu_plugin ) def get_file_info(path): file_stat = os.stat(path) # Get file permissions file_permissions = stat.filemode(file_stat.st_mode) # Get number of links num_links = file_stat.st_nlink # Get owner name uid = file_stat.st_uid owner_name = pwd.getpwuid(uid)[0] # Get group name gid = file_stat.st_gid group_name = grp.getgrgid(gid)[0] # Get file size file_size = file_stat.st_size # Get last modification time mod_time = time.strftime('%b %d %H:%M', time.localtime(file_stat.st_mtime)) return f"{file_permissions} {num_links} {owner_name} {group_name} {file_size} {mod_time}" def _gather_plugin_version(file_path): version = '' if os.path.isfile(file_path): with open(file_path, 'r') as file: content = file.read() match = re.search(r'Version:\s+(.)', content) if match: version = match.group(1) return version def plugin_installed(mu_plugin_dir): plugin_must_use_file = os.path.join(mu_plugin_dir, 'cl-smart-advice.php') plugin_must_use_folder = os.path.join(mu_plugin_dir, 'cl-smart-advice') return True if os.path.isfile(plugin_must_use_file) and os.path.isdir(plugin_must_use_folder) else False def plugin_cleanup(mu_plugin_dir): if check_for_root(): raise PermissionError('Root privileges are not allowed to run plugin_cleanup method.') plugin_must_use_file = os.path.join(mu_plugin_dir, 'cl-smart-advice.php') plugin_must_use_folder = os.path.join(mu_plugin_dir, 'cl-smart-advice') # Removing the old version of the plugin installation via a symlink if os.path.islink(plugin_must_use_file): os.remove(plugin_must_use_file) logging.info('Remove outdated plugin symlink in %s.', mu_plugin_dir) # The order of deletion matters! if os.path.exists(plugin_must_use_file): os.remove(plugin_must_use_file) if os.path.exists(plugin_must_use_folder): shutil.rmtree(plugin_must_use_folder) def setup_plugin(mu_plugin_dir): base_dir = '/opt/alt/php-xray/php/smart-advice-plugin/' plugin_file = os.path.join(base_dir, 'cl-smart-advice.php') plugin_folder = os.path.join(base_dir, 'cl-smart-advice') plugin_version = _gather_plugin_version(plugin_file) plugin_must_use_file = os.path.join(mu_plugin_dir, 'cl-smart-advice.php') plugin_must_use_folder = os.path.join(mu_plugin_dir, 'cl-smart-advice') plugin_must_use_version = _gather_plugin_version(plugin_must_use_file) if plugin_installed(mu_plugin_dir) and plugin_version == plugin_must_use_version: logging.info('Smart advice plugin %s is already installed in %s, nothing to do.', plugin_version, mu_plugin_dir) return # Remove outdated items plugin_cleanup(mu_plugin_dir) try: # The copy order matters! shutil.copytree(plugin_folder, plugin_must_use_folder) shutil.copy2(plugin_file, plugin_must_use_file) except OSError as e: folder_info = get_file_info(mu_plugin_dir) logging.error("Failed copy plugin files. (Folder info: %s). Reason: %s}", folder_info, str(e)) raise def install_plugin(wp_site: WebsiteInfo): setup_plugin(wp_site.wp_must_use_plugins_path) def uninstall_plugin(args: Args, wp_site: WebsiteInfo): mu_plugin_dir = wp_site.wp_must_use_plugins_path if not plugin_installed(mu_plugin_dir): logging.info('Smart advice plugin is not installed in %s, nothing to do.', mu_plugin_dir) return try: uninstall_result = subprocess.check_output([ get_wp_cli_command(args), 'smart-advice', 'uninstall'], text=True, stderr=subprocess.PIPE, env=_get_wp_cli_environment(), ).strip() plugin_cleanup(mu_plugin_dir) except subprocess.CalledProcessError as e: logging.error(f"Error happened while uninstalling WP SmartAdvice plugin" f"\nstdout=`%s`\nstderr=`%s`,\nreturncode=%s" '', e.stdout.strip('\n'), e.stderr.strip('\n'), e.returncode) raise def sync_advice(args: Args, wp_site: WebsiteInfo, data: str): try: sync_result = subprocess.check_output( [get_wp_cli_command(args), 'smart-advice', 'sync', '--list=' + data, '--cpanel_url=' + args.login_url, '--cpanel_user_emails=' + args.emails], text=True, stderr=subprocess.PIPE, env=_get_wp_cli_environment() ).strip() except subprocess.CalledProcessError as e: logging.error(f"Failed run wp-cli command smart-advice sync." f"\nstdout=`%s`\nstderr=`%s`,\nreturncode=%s" '', e.stdout.strip('\n'), e.stderr.strip('\n'), e.returncode) _report_sync_or_error(args, 'advice_sync_failed') raise else: _report_sync_or_error(args, 'advice_synced') def health_check(args: Args, wp_site: WebsiteInfo): try: result = subprocess.check_output([*get_wp_cli_command(args), 'option', 'get', 'home'], env=_get_wp_cli_environment() ).decode().strip() except subprocess.CalledProcessError as e: uninstall_plugin(args, wp_site) logging.error(f"Failed run wp-cli command, site's health status failed. Plugin removed.") raise def main(args: Args): wordpress_website = gather_information(args) if not wordpress_website: exit(1) if args.action == Action.UNINSTALL: uninstall_plugin(args, wordpress_website) else: install_plugin(wordpress_website) sync_advice(args, wordpress_website, data=args.advice_list_json) health_check(args, wordpress_website) if __name__ == '__main__': sentry_init() # FIXME: migrate to proper argparse one day def get_action(): if len(sys.argv) >= 8 and sys.argv[7] == 'true': return Action.UNINSTALL return Action.INSTALL args = Args( path_to_php=sys.argv[1], path_to_wp=sys.argv[2], path_to_mu_plugin=sys.argv[3], advice_list_json=sys.argv[4], emails=sys.argv[5], login_url=sys.argv[6], action=get_action() ) logging.info('Arguments received: %s', args) main(args)

| ver. 1.4 | Github | . | PHP 8.1.33 | Генерация страницы: 5.52 | proxy | phpinfo | Настройка