✏️ Editing: smart_advice_plugin_manager.py Read Only

#!/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 _validate_mu_plugin_dir(mu_plugin_dir):
    """Verify resolved mu_plugin_dir parent is owned by the current user.

Uses ownership instead of path prefix — works on any panel layout.
    Runs under cagefs_enter_user / sudo -u, so os.getuid() == site owner.
    """
    if not mu_plugin_dir:
        return False
    resolved = os.path.realpath(mu_plugin_dir)
    parent = os.path.dirname(resolved)
    try:
        parent_stat = os.lstat(parent)
    except OSError:
        logging.error('Cannot stat parent of mu_plugin_dir: %s', parent)
        return False
    if parent_stat.st_uid != os.getuid():
        logging.error('mu_plugin_dir parent %s owned by uid=%s, expected uid=%s',
                      parent, parent_stat.st_uid, os.getuid())
        return False
    return True

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.')
    if not _validate_mu_plugin_dir(mu_plugin_dir):
        raise ValueError('mu_plugin_dir failed path validation: %s' % 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')

# 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):
    if not _validate_mu_plugin_dir(mu_plugin_dir):
        raise ValueError('mu_plugin_dir failed path validation: %s' % 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)

🔐 Sid Gifari File Manager Pro